untangle input boolean expressions from models (#460)

<!-- Thank you for submitting this PR! :) -->

## Description

Previously boolean expressions were only used on where clauses for
models. We'd also like to use them for arguments for commands to make
permissions work. This PR splits a boolean expression from it's model.
This has the nice side effect of allowing the same boolean expression
type to used across multiple models, which is a sensible thing to want
to be able to do.

Before this change, using the same boolean expression type on two models
would give you this error:

<img width="783" alt="Screenshot 2024-04-19 at 11 53 23"
src="https://github.com/hasura/v3-engine/assets/4729125/bcc7a4b9-8b6f-4d82-9860-190621c0f5fd">

<!--
  Questions to consider answering:
  1. What user-facing changes are being made?
2. What are issues related to this PR? (Consider adding `(close
#<issue-no>)` to the PR title)
  3. What is the conceptual design behind this PR?
  4. How can this PR be tested/verified?
  5. Does the PR have limitations?
  6. Does the PR introduce breaking changes?
-->

## Changelog

- Add a changelog entry (in the "Changelog entry" section below) if the
changes in this PR have any user-facing impact. See [changelog
guide](https://github.com/hasura/graphql-engine-mono/wiki/Changelog-Guide).
- If no changelog is required ignore/remove this section and add a
`no-changelog-required` label to the PR.

### Product
_(Select all products this will be available in)_
- [X] community-edition
- [X] cloud
<!-- product : end : DO NOT REMOVE -->

### Type
<!-- See changelog structure:
https://github.com/hasura/graphql-engine-mono/wiki/Changelog-Guide#structure-of-our-changelog
-->
_(Select only one. In case of multiple, choose the most appropriate)_
- [ ] highlight
- [X] enhancement
- [ ] bugfix
- [ ] behaviour-change
- [ ] performance-enhancement
- [ ] security-fix
<!-- type : end : DO NOT REMOVE -->

### Changelog entry
<!--
  - Add a user understandable changelog entry
- Include all details needed to understand the change. Try including
links to docs or issues if relevant
  - For Highlights start with a H4 heading (#### <entry title>)
  - Get the changelog entry reviewed by your team
-->

Allow the same `ObjectBooleanExpressionType` to be shared between
multiple models.

<!-- changelog-entry : end : DO NOT REMOVE -->

<!-- changelog : end : DO NOT REMOVE -->

V3_GIT_ORIGIN_REV_ID: 6c9979ddaad50d476c0996d1ece48f0cf1c8e99d

v3/crates/engine/src/execute/ir/filter.rs

@@ -10,7 +10,7 @@ use crate::execute::error;
 use crate::execute::model_tracking::{count_model, UsagesCounts};
 use crate::schema::types::output_type::relationship::FilterRelationshipAnnotation;
 use crate::schema::types::{self};
-use crate::schema::types::{InputAnnotation, ModelInputAnnotation};
+use crate::schema::types::{BooleanExpressionAnnotation, InputAnnotation, ModelInputAnnotation};
 use crate::schema::GDS;
 
 use super::relationship::LocalModelRelationshipInfo;
@@ -58,8 +58,8 @@ pub(crate) fn build_filter_expression<'s>(
 ) -> Result<Vec<ndc_models::Expression>, error::Error> {
     match field.info.generic {
         // "_and"
-        types::Annotation::Input(InputAnnotation::Model(
+        types::Annotation::Input(InputAnnotation::BooleanExpression(
-            ModelInputAnnotation::ModelFilterArgument {
+            BooleanExpressionAnnotation::BooleanExpressionArgument {
                 field: types::ModelFilterArgument::AndOp,
             },
         )) => {
@@ -78,8 +78,8 @@ pub(crate) fn build_filter_expression<'s>(
             Ok(vec![expression])
         }
         // "_or"
-        types::Annotation::Input(InputAnnotation::Model(
+        types::Annotation::Input(InputAnnotation::BooleanExpression(
-            ModelInputAnnotation::ModelFilterArgument {
+            BooleanExpressionAnnotation::BooleanExpressionArgument {
                 field: types::ModelFilterArgument::OrOp,
             },
         )) => {
@@ -98,8 +98,8 @@ pub(crate) fn build_filter_expression<'s>(
             Ok(vec![expression])
         }
         // "_not"
-        types::Annotation::Input(InputAnnotation::Model(
+        types::Annotation::Input(InputAnnotation::BooleanExpression(
-            ModelInputAnnotation::ModelFilterArgument {
+            BooleanExpressionAnnotation::BooleanExpressionArgument {
                 field: types::ModelFilterArgument::NotOp,
             },
         )) => {
@@ -120,8 +120,8 @@ pub(crate) fn build_filter_expression<'s>(
         // to be a relationship column, we'll have to join all the paths to
         // specify NDC, what relationships needs to be traversed to access this
         // column. The order decides how to access the column.
-        types::Annotation::Input(InputAnnotation::Model(
+        types::Annotation::Input(InputAnnotation::BooleanExpression(
-            ModelInputAnnotation::ModelFilterArgument {
+            BooleanExpressionAnnotation::BooleanExpressionArgument {
                 field: types::ModelFilterArgument::Field { ndc_column: column },
             },
         )) => {
@@ -158,8 +158,8 @@ pub(crate) fn build_filter_expression<'s>(
         }
         // Relationship field used for filtering.
         // This relationship can either point to another relationship or a column.
-        types::Annotation::Input(InputAnnotation::Model(
+        types::Annotation::Input(InputAnnotation::BooleanExpression(
-            ModelInputAnnotation::ModelFilterArgument {
+            BooleanExpressionAnnotation::BooleanExpressionArgument {
                 field:
                     types::ModelFilterArgument::RelationshipField(FilterRelationshipAnnotation {
                         relationship_name,

v3/crates/engine/src/execute/ir/query_root/select_many.rs

@@ -22,8 +22,7 @@ use crate::execute::ir::permissions;
 use crate::execute::model_tracking::{count_model, UsagesCounts};
 use crate::metadata::resolved;
 use crate::metadata::resolved::subgraph::Qualified;
-
+use crate::schema::types::{self, Annotation, BooleanExpressionAnnotation, ModelInputAnnotation};
-use crate::schema::types::{self, Annotation, ModelInputAnnotation};
 use crate::schema::GDS;
 
 /// IR for the 'select_many' operation on a model
@@ -85,12 +84,6 @@ pub(crate) fn select_many_generate_ir<'n, 's>(
                             .map_err(error::Error::map_unexpected_value_to_external_error)?,
                     )
                 }
-                ModelInputAnnotation::ModelFilterExpression => {
-                    filter_clause = filter::resolve_filter_expression(
-                        argument.value.as_object()?,
-                        &mut usage_counts,
-                    )?;
-                }
                 ModelInputAnnotation::ModelArgumentsExpression => match &argument.value {
                     normalized_ast::Value::Object(arguments) => {
                         model_arguments.extend(
@@ -116,6 +109,15 @@ pub(crate) fn select_many_generate_ir<'n, 's>(
                 }
             },
 
+            Annotation::Input(types::InputAnnotation::BooleanExpression(
+                BooleanExpressionAnnotation::BooleanExpression,
+            )) => {
+                filter_clause = filter::resolve_filter_expression(
+                    argument.value.as_object()?,
+                    &mut usage_counts,
+                )?;
+            }
+
             annotation => {
                 return Err(error::InternalEngineError::UnexpectedAnnotation {
                     annotation: annotation.clone(),

v3/crates/engine/src/execute/ir/relationship.rs

@@ -39,7 +39,7 @@ use crate::{
 use crate::{
     metadata::resolved::{self, subgraph::Qualified},
     schema::{
-        types::{Annotation, InputAnnotation, ModelInputAnnotation},
+        types::{Annotation, BooleanExpressionAnnotation, InputAnnotation, ModelInputAnnotation},
         GDS,
     },
 };
@@ -248,12 +248,6 @@ pub(crate) fn generate_model_relationship_ir<'s>(
                                     error::Error::map_unexpected_value_to_external_error,
                                 )?)
                             }
-                            ModelInputAnnotation::ModelFilterExpression => {
-                                filter_clause = resolve_filter_expression(
-                                    argument.value.as_object()?,
-                                    usage_counts,
-                                )?
-                            }
                             ModelInputAnnotation::ModelOrderByExpression => {
                                 order_by = Some(build_ndc_order_by(argument, usage_counts)?)
                             }
@@ -264,6 +258,13 @@ pub(crate) fn generate_model_relationship_ir<'s>(
                             }
                         }
                     }
+                    InputAnnotation::BooleanExpression(
+                        BooleanExpressionAnnotation::BooleanExpression,
+                    ) => {
+                        filter_clause =
+                            resolve_filter_expression(argument.value.as_object()?, usage_counts)?
+                    }
+
                     _ => {
                         return Err(error::InternalEngineError::UnexpectedAnnotation {
                             annotation: annotation.clone(),

v3/crates/engine/src/metadata/resolved.rs

@@ -1,4 +1,5 @@
 mod argument;
+pub mod boolean_expression;
 pub mod command;
 pub mod data_connector;
 pub mod error;

v3/crates/engine/src/metadata/resolved/boolean_expression.rs

@@ -0,0 +1,155 @@
+use super::stages::data_connector_scalar_types;
+
+use crate::metadata::resolved::data_connector;
+
+use crate::metadata::resolved::error::{BooleanExpressionError, Error, GraphqlConfigError};
+use crate::metadata::resolved::model;
+
+use crate::metadata::resolved::stages::graphql_config::GraphqlConfig;
+use crate::metadata::resolved::subgraph::{Qualified, QualifiedTypeReference};
+
+use crate::metadata::resolved::types::TypeMapping;
+
+use lang_graphql::ast::common::{self as ast};
+use ndc_models;
+
+use open_dds::{
+    data_connector::DataConnectorName,
+    types::{CustomTypeName, FieldName},
+};
+use serde::{Deserialize, Serialize};
+use std::collections::{BTreeMap, HashMap};
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct ComparisonExpressionInfo {
+    pub data_connector_name: Qualified<DataConnectorName>,
+    pub scalar_type_name: String,
+    pub type_name: ast::TypeName,
+    pub ndc_column: String,
+    pub operators: BTreeMap<String, QualifiedTypeReference>,
+    pub is_null_operator_name: String,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct BooleanExpressionGraphqlConfig {
+    pub where_field_name: ast::Name,
+    pub and_operator_name: ast::Name,
+    pub or_operator_name: ast::Name,
+    pub not_operator_name: ast::Name,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct BooleanExpression {
+    pub type_name: ast::TypeName,
+    pub scalar_fields: HashMap<FieldName, ComparisonExpressionInfo>,
+    pub graphql_config: BooleanExpressionGraphqlConfig,
+}
+
+// record filter expression info
+pub fn resolve_boolean_expression(
+    name: &Qualified<CustomTypeName>,
+    data_connector_name: &Qualified<DataConnectorName>,
+    where_type_name: ast::TypeName,
+    subgraph: &str,
+    data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
+    type_mappings: &TypeMapping,
+    graphql_config: &GraphqlConfig,
+) -> Result<BooleanExpression, Error> {
+    let mut scalar_fields = HashMap::new();
+
+    let scalar_types = &data_connectors
+        .data_connectors_with_scalars
+        .get(data_connector_name)
+        .ok_or(Error::BooleanExpressionError {
+            boolean_expression_error:
+                BooleanExpressionError::UnknownDataConnectorInObjectBooleanExpressionType {
+                    boolean_expression_type: name.clone(),
+                    data_connector: data_connector_name.clone(),
+                },
+        })?
+        .scalars;
+
+    let TypeMapping::Object { field_mappings, .. } = type_mappings;
+
+    let filter_graphql_config = graphql_config
+        .query
+        .filter_input_config
+        .as_ref()
+        .ok_or_else(|| Error::GraphqlConfigError {
+            graphql_config_error: GraphqlConfigError::MissingFilterInputFieldInGraphqlConfig,
+        })?;
+
+    for (field_name, field_mapping) in field_mappings.iter() {
+        // Generate comparison expression for fields mapped to simple scalar type
+        if let Some((scalar_type_name, scalar_type_info)) =
+            data_connector::get_simple_scalar(field_mapping.column_type.clone(), scalar_types)
+        {
+            if let Some(graphql_type_name) = &scalar_type_info.comparison_expression_name.clone() {
+                let mut operators = BTreeMap::new();
+                for (op_name, op_definition) in
+                    scalar_type_info.scalar_type.comparison_operators.iter()
+                {
+                    operators.insert(
+                        op_name.clone(),
+                        model::resolve_ndc_type(
+                            data_connector_name,
+                            &get_argument_type(op_definition, &field_mapping.column_type),
+                            scalar_types,
+                            subgraph,
+                        )?,
+                    );
+                }
+
+                // Register scalar comparison field only if it contains non-zero operators.
+                if !operators.is_empty() {
+                    scalar_fields.insert(
+                        field_name.clone(),
+                        ComparisonExpressionInfo {
+                            data_connector_name: data_connector_name.clone(),
+                            scalar_type_name: scalar_type_name.clone(),
+                            type_name: graphql_type_name.clone(),
+                            ndc_column: field_mapping.column.clone(),
+                            operators,
+                            is_null_operator_name: filter_graphql_config
+                                .operator_names
+                                .is_null
+                                .to_string(),
+                        },
+                    );
+                };
+            }
+        }
+    }
+
+    Ok(BooleanExpression {
+        type_name: where_type_name,
+        scalar_fields,
+        graphql_config: (BooleanExpressionGraphqlConfig {
+            where_field_name: filter_graphql_config.where_field_name.clone(),
+            and_operator_name: filter_graphql_config.operator_names.and.clone(),
+            or_operator_name: filter_graphql_config.operator_names.or.clone(),
+            not_operator_name: filter_graphql_config.operator_names.not.clone(),
+        }),
+    })
+}
+
+fn unwrap_nullable(field_type: &ndc_models::Type) -> &ndc_models::Type {
+    if let ndc_models::Type::Nullable { underlying_type } = field_type {
+        unwrap_nullable(underlying_type)
+    } else {
+        field_type
+    }
+}
+
+fn get_argument_type(
+    op_definition: &ndc_models::ComparisonOperatorDefinition,
+    field_type: &ndc_models::Type,
+) -> ndc_models::Type {
+    match op_definition {
+        ndc_models::ComparisonOperatorDefinition::Equal => unwrap_nullable(field_type).clone(),
+        ndc_models::ComparisonOperatorDefinition::In => ndc_models::Type::Array {
+            element_type: Box::new(unwrap_nullable(field_type).clone()),
+        },
+        ndc_models::ComparisonOperatorDefinition::Custom { argument_type } => argument_type.clone(),
+    }
+}

v3/crates/engine/src/metadata/resolved/error.rs

@@ -605,6 +605,11 @@ pub enum BooleanExpressionError {
         data_connector_object_type: String,
         data_connector: Qualified<DataConnectorName>,
     },
+    #[error("{error:} in boolean expression type {boolean_expression_type:}")]
+    BooleanExpressionTypeMappingCollectionError {
+        boolean_expression_type: Qualified<CustomTypeName>,
+        error: TypeMappingCollectionError,
+    },
     #[error("the following object boolean expression type is defined more than once: {name:}")]
     DuplicateObjectBooleanExpressionTypeDefinition { name: Qualified<CustomTypeName> },
     #[error("unknown object boolean expression type {name:} is used in model {model:}")]

v3/crates/engine/src/metadata/resolved/metadata.rs

@@ -63,7 +63,9 @@ pub fn resolve_metadata(
         data_connectors,
         data_connector_type_mappings,
         &object_types,
+        scalar_types,
         &mut existing_graphql_types,
+        graphql_config,
     )?;
 
     // resolve models
@@ -252,7 +254,9 @@ fn resolve_boolean_expression_types(
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
     data_connector_type_mappings: &data_connector_type_mappings::DataConnectorTypeMappings,
     object_types: &HashMap<Qualified<CustomTypeName>, ObjectTypeRepresentation>,
+    scalar_types: &HashMap<Qualified<CustomTypeName>, scalar_types::ScalarTypeRepresentation>,
     existing_graphql_types: &mut HashSet<ast::TypeName>,
+    graphql_config: &graphql_config::GraphqlConfig,
 ) -> Result<HashMap<Qualified<CustomTypeName>, ObjectBooleanExpressionType>, Error> {
     let mut boolean_expression_types = HashMap::new();
     for open_dds::accessor::QualifiedObject {
@@ -264,9 +268,11 @@ fn resolve_boolean_expression_types(
             boolean_expression_type,
             subgraph,
             data_connectors,
-            object_types,
             data_connector_type_mappings,
+            object_types,
+            scalar_types,
             existing_graphql_types,
+            graphql_config,
         )?;
         if let Some(existing) = boolean_expression_types.insert(
             resolved_boolean_expression.name.clone(),
@@ -347,7 +353,6 @@ fn resolve_models(
             resolve_model_graphql_api(
                 model_graphql_definition,
                 &mut resolved_model,
-                subgraph,
                 existing_graphql_types,
                 data_connectors,
                 &model.description,
@@ -398,6 +403,7 @@ fn resolve_relationships(
             data_connectors,
             object_representation,
         )?;
+
         if object_representation
             .relationships
             .insert(

v3/crates/engine/src/metadata/resolved/model.rs

@@ -7,6 +7,7 @@ use super::types::{
     ObjectTypeRepresentation, TypeMappingToCollect,
 };
 use crate::metadata::resolved::argument::get_argument_mappings;
+
 use crate::metadata::resolved::data_connector;
 use crate::metadata::resolved::data_connector::DataConnectorLink;
 use crate::metadata::resolved::error::{
@@ -63,31 +64,6 @@ pub struct SelectManyGraphQlDefinition {
     pub deprecated: Option<Deprecated>,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct ComparisonExpressionInfo {
-    pub data_connector_name: Qualified<DataConnectorName>,
-    pub scalar_type_name: String,
-    pub type_name: ast::TypeName,
-    pub ndc_column: String,
-    pub operators: BTreeMap<String, QualifiedTypeReference>,
-    pub is_null_operator_name: String,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct ModelFilterExpressionGraphqlConfig {
-    pub where_field_name: ast::Name,
-    pub and_operator_name: ast::Name,
-    pub or_operator_name: ast::Name,
-    pub not_operator_name: ast::Name,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct ModelFilterExpression {
-    pub where_type_name: ast::TypeName,
-    pub scalar_fields: HashMap<FieldName, ComparisonExpressionInfo>,
-    pub filter_graphql_config: ModelFilterExpressionGraphqlConfig,
-}
-
 // TODO: add support for aggregates
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct OrderByExpressionInfo {
@@ -121,7 +97,6 @@ pub struct ModelGraphQlApi {
     pub arguments_input_config: Option<ModelGraphqlApiArgumentsConfig>,
     pub select_uniques: Vec<SelectUniqueGraphQlDefinition>,
     pub select_many: Option<SelectManyGraphQlDefinition>,
-    pub filter_expression: Option<ModelFilterExpression>,
     pub order_by_expression: Option<ModelOrderByExpression>,
     pub limit_field: Option<LimitFieldGraphqlConfig>,
     pub offset_field: Option<OffsetFieldGraphqlConfig>,
@@ -416,7 +391,7 @@ pub fn resolve_model(
 }
 
 // helper function to resolve ndc types to dds type based on scalar type representations
-fn resolve_ndc_type(
+pub(crate) fn resolve_ndc_type(
     data_connector: &Qualified<DataConnectorName>,
     source_type: &ndc_models::Type,
     scalars: &HashMap<&str, data_connector_scalar_types::ScalarTypeWithRepresentationInfo>,
@@ -939,7 +914,6 @@ pub(crate) fn get_ndc_column_for_comparison<F: Fn() -> String>(
 pub fn resolve_model_graphql_api(
     model_graphql_definition: &ModelGraphQlDefinition,
     model: &mut Model,
-    subgraph: &str,
     existing_graphql_types: &mut HashSet<ast::TypeName>,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
     model_description: &Option<String>,
@@ -1063,120 +1037,6 @@ pub fn resolve_model_graphql_api(
         .transpose()?
         .flatten();
 
-    // record filter expression info
-    model.graphql_api.filter_expression = model
-        .filter_expression_type
-        .as_ref()
-        .map(
-            |model_filter_expression_type| -> Result<Option<ModelFilterExpression>, Error> {
-                let model_source = model.source.as_ref().ok_or_else(|| {
-                    Error::CannotUseFilterExpressionsWithoutSource {
-                        model: model.name.clone(),
-                    }
-                })?;
-                let where_type_name = if let Some(model_filter_expression_graphql) =
-                    &model_filter_expression_type.graphql
-                {
-                    model_filter_expression_graphql.type_name.clone()
-                } else {
-                    return Ok(None);
-                };
-                let mut scalar_fields = HashMap::new();
-
-                let scalar_types = &data_connectors
-                    .data_connectors_with_scalars
-                    .get(&model_source.data_connector.name)
-                    .ok_or(Error::UnknownModelDataConnector {
-                        model_name: model_name.clone(),
-                        data_connector: model_source.data_connector.name.clone(),
-                    })?
-                    .scalars;
-
-                let TypeMapping::Object { field_mappings, .. } = model_source
-                    .type_mappings
-                    .get(&model.data_type)
-                    .ok_or(Error::TypeMappingRequired {
-                        model_name: model_name.clone(),
-                        type_name: model.data_type.clone(),
-                        data_connector: model_source.data_connector.name.clone(),
-                    })?;
-
-                let filter_graphql_config = graphql_config
-                    .query
-                    .filter_input_config
-                    .as_ref()
-                    .ok_or_else(|| Error::GraphqlConfigError {
-                        graphql_config_error:
-                            GraphqlConfigError::MissingFilterInputFieldInGraphqlConfig,
-                    })?;
-
-                for (field_name, field_mapping) in field_mappings.iter() {
-                    // Generate comparison expression for fields mapped to simple scalar type
-                    if let Some((scalar_type_name, scalar_type_info)) =
-                        data_connector::get_simple_scalar(
-                            field_mapping.column_type.clone(),
-                            scalar_types,
-                        )
-                    {
-                        if let Some(graphql_type_name) =
-                            &scalar_type_info.comparison_expression_name.clone()
-                        {
-                            let mut operators = BTreeMap::new();
-                            for (op_name, op_definition) in
-                                scalar_type_info.scalar_type.comparison_operators.iter()
-                            {
-                                operators.insert(
-                                    op_name.clone(),
-                                    resolve_ndc_type(
-                                        &model_source.data_connector.name,
-                                        &get_argument_type(
-                                            op_definition,
-                                            &field_mapping.column_type,
-                                        ),
-                                        scalar_types,
-                                        subgraph,
-                                    )?,
-                                );
-                            }
-                            // Register scalar comparison field only if it contains non-zero operators.
-                            if !operators.is_empty() {
-                                scalar_fields.insert(
-                                    field_name.clone(),
-                                    ComparisonExpressionInfo {
-                                        data_connector_name: model_source
-                                            .data_connector
-                                            .name
-                                            .clone(),
-                                        scalar_type_name: scalar_type_name.clone(),
-                                        type_name: graphql_type_name.clone(),
-                                        ndc_column: field_mapping.column.clone(),
-                                        operators,
-                                        is_null_operator_name: filter_graphql_config
-                                            .operator_names
-                                            .is_null
-                                            .to_string(),
-                                    },
-                                );
-                            };
-                        }
-                    }
-                }
-
-                Ok(Some(ModelFilterExpression {
-                    where_type_name,
-                    scalar_fields,
-                    filter_graphql_config: (ModelFilterExpressionGraphqlConfig {
-                        where_field_name: filter_graphql_config.where_field_name.clone(),
-                        and_operator_name: filter_graphql_config.operator_names.and.clone(),
-                        or_operator_name: filter_graphql_config.operator_names.or.clone(),
-                        not_operator_name: filter_graphql_config.operator_names.not.clone(),
-                    }),
-                }))
-            },
-        )
-        .transpose()?
-        .flatten();
-
     // record select_many root field
     model.graphql_api.select_many = match &model_graphql_definition.select_many {
         None => Ok(None),
@@ -1250,27 +1110,6 @@ pub fn resolve_model_graphql_api(
     Ok(())
 }
 
-fn unwrap_nullable(field_type: &ndc_models::Type) -> &ndc_models::Type {
-    if let ndc_models::Type::Nullable { underlying_type } = field_type {
-        unwrap_nullable(underlying_type)
-    } else {
-        field_type
-    }
-}
-
-fn get_argument_type(
-    op_definition: &ndc_models::ComparisonOperatorDefinition,
-    field_type: &ndc_models::Type,
-) -> ndc_models::Type {
-    match op_definition {
-        ndc_models::ComparisonOperatorDefinition::Equal => unwrap_nullable(field_type).clone(),
-        ndc_models::ComparisonOperatorDefinition::In => ndc_models::Type::Array {
-            element_type: Box::new(unwrap_nullable(field_type).clone()),
-        },
-        ndc_models::ComparisonOperatorDefinition::Custom { argument_type } => argument_type.clone(),
-    }
-}
-
 pub fn resolve_model_source(
     model_source: &models::ModelSource,
     model: &mut Model,

v3/crates/engine/src/metadata/resolved/stages/data_connectors/types.rs

@@ -51,6 +51,7 @@ pub struct ComparisonOperators {
 }
 
 // basic scalar type info
+#[derive(Debug)]
 pub struct ScalarTypeInfo<'a> {
     pub scalar_type: &'a ndc_models::ScalarType,
     pub comparison_operators: ComparisonOperators,

v3/crates/engine/src/metadata/resolved/types.rs

@@ -1,10 +1,15 @@
+use super::stages::{
+    data_connector_scalar_types, data_connector_type_mappings, graphql_config, scalar_types,
+};
+use crate::metadata::resolved::boolean_expression;
+use crate::metadata::resolved::data_connector;
 use crate::metadata::resolved::error::{BooleanExpressionError, Error};
+
 use crate::metadata::resolved::relationship::Relationship;
 use crate::metadata::resolved::subgraph::{
     mk_qualified_type_reference, Qualified, QualifiedBaseType, QualifiedTypeName,
     QualifiedTypeReference,
 };
-
 use indexmap::IndexMap;
 use lang_graphql::ast::common as ast;
 use ndc_models;
@@ -22,8 +27,6 @@ use std::collections::{BTreeMap, HashMap, HashSet};
 use std::str::FromStr;
 
 use super::ndc_validation::{get_underlying_named_type, NDCValidationError};
-use super::stages::data_connector_scalar_types;
-use super::stages::{data_connector_type_mappings, scalar_types};
 use super::typecheck;
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, derive_more::Display)]
@@ -83,13 +86,10 @@ pub struct ObjectBooleanExpressionType {
     pub name: Qualified<CustomTypeName>,
     pub object_type: Qualified<CustomTypeName>,
     pub data_connector_name: Qualified<DataConnectorName>,
+    pub data_connector_link: data_connector::DataConnectorLink,
     pub data_connector_object_type: String,
-    pub graphql: Option<ObjectBooleanExpressionTypeGraphQlConfiguration>,
+    pub type_mappings: BTreeMap<Qualified<types::CustomTypeName>, TypeMapping>,
-}
+    pub graphql: Option<boolean_expression::BooleanExpression>,
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct ObjectBooleanExpressionTypeGraphQlConfiguration {
-    pub type_name: ast::TypeName,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, derive_more::Display)]
@@ -433,9 +433,11 @@ pub(crate) fn resolve_object_boolean_expression_type(
     object_boolean_expression: &ObjectBooleanExpressionTypeV1,
     subgraph: &str,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-    object_types: &HashMap<Qualified<CustomTypeName>, ObjectTypeRepresentation>,
     data_connector_type_mappings: &data_connector_type_mappings::DataConnectorTypeMappings,
+    object_types: &HashMap<Qualified<CustomTypeName>, ObjectTypeRepresentation>,
+    scalar_types: &HashMap<Qualified<CustomTypeName>, scalar_types::ScalarTypeRepresentation>,
     existing_graphql_types: &mut HashSet<ast::TypeName>,
+    graphql_config: &graphql_config::GraphqlConfig,
 ) -> Result<ObjectBooleanExpressionType, Error> {
     // name of the boolean expression
     let qualified_name = Qualified::new(
@@ -544,26 +546,88 @@ pub(crate) fn resolve_object_boolean_expression_type(
                 });
     }
 
+    let boolean_expression_type =
+        Qualified::new(subgraph.to_string(), object_boolean_expression.name.clone());
+
+    let object_type = Qualified::new(
+        subgraph.to_string(),
+        object_boolean_expression.object_type.clone(),
+    );
+
+    let data_connector_name = Qualified::new(
+        subgraph.to_string(),
+        object_boolean_expression.data_connector_name.clone(),
+    );
+
+    // Collect type mappings.
+    let mut type_mappings = BTreeMap::new();
+
+    let type_mapping_to_collect = TypeMappingToCollect {
+        type_name: &object_type,
+        ndc_object_type_name: object_boolean_expression
+            .data_connector_object_type
+            .as_str(),
+    };
+    collect_type_mapping_for_source(
+        &type_mapping_to_collect,
+        data_connector_type_mappings,
+        &qualified_data_connector_name,
+        object_types,
+        scalar_types,
+        &mut type_mappings,
+    )
+    .map_err(|error| {
+        Error::from(
+            BooleanExpressionError::BooleanExpressionTypeMappingCollectionError {
+                boolean_expression_type: boolean_expression_type.clone(),
+                error,
+            },
+        )
+    })?;
+
     // validate graphql config
-    let graphql_config = object_boolean_expression
+    let boolean_expression_graphql_config = object_boolean_expression
         .graphql
         .as_ref()
-        .map(|graphql_config| {
+        .map(|object_boolean_graphql_config| {
             let graphql_type_name =
-                mk_name(graphql_config.type_name.0.as_ref()).map(ast::TypeName)?;
+                mk_name(object_boolean_graphql_config.type_name.0.as_ref()).map(ast::TypeName)?;
+
             store_new_graphql_type(existing_graphql_types, Some(&graphql_type_name))?;
-            Ok::<_, Error>(ObjectBooleanExpressionTypeGraphQlConfiguration {
+
-                type_name: graphql_type_name,
+            let type_mapping = type_mappings
-            })
+                .get(&Qualified::new(
+                    subgraph.to_string(),
+                    object_boolean_expression.object_type.clone(),
+                ))
+                .unwrap();
+
+            boolean_expression::resolve_boolean_expression(
+                &boolean_expression_type,
+                &data_connector_name,
+                graphql_type_name.clone(),
+                subgraph,
+                data_connectors,
+                type_mapping,
+                graphql_config,
+            )
         })
         .transpose()?;
 
+    let data_connector_link = data_connector::DataConnectorLink::new(
+        data_connector_name,
+        data_connector_context.inner.url.clone(),
+        data_connector_context.inner.headers,
+    )?;
+
     let resolved_boolean_expression = ObjectBooleanExpressionType {
         name: qualified_name.clone(),
+        type_mappings,
         object_type: qualified_object_type_name.clone(),
         data_connector_name: qualified_data_connector_name,
+        data_connector_link,
         data_connector_object_type: object_boolean_expression.data_connector_object_type.clone(),
-        graphql: graphql_config,
+        graphql: boolean_expression_graphql_config,
     };
     Ok(resolved_boolean_expression)
 }

v3/crates/engine/src/schema.rs

@@ -20,6 +20,7 @@ use crate::metadata::{
 use self::types::{PossibleApolloFederationTypes, RootFieldAnnotation};
 
 pub mod apollo_federation;
+pub mod boolean_expression;
 pub mod commands;
 pub mod model_arguments;
 pub mod model_filter;
@@ -106,6 +107,15 @@ impl gql_schema::SchemaContext for GDS {
                 gds_type_name,
                 graphql_type_name,
             ),
+            types::TypeId::InputObjectBooleanExpressionType {
+                gds_type_name,
+                graphql_type_name,
+            } => boolean_expression::build_boolean_expression_input_schema(
+                self,
+                builder,
+                graphql_type_name,
+                gds_type_name,
+            ),
             types::TypeId::NodeRoot => Ok(gql_schema::TypeInfo::Interface(
                 relay::node_interface_schema(builder, self)?,
             )),
@@ -115,15 +125,6 @@ impl gql_schema::SchemaContext for GDS {
             } => model_arguments::build_model_arguments_input_schema(
                 self, builder, type_name, model_name,
             ),
-            types::TypeId::ModelBooleanExpression {
-                model_name,
-                graphql_type_name,
-            } => model_filter::build_model_filter_expression_input_schema(
-                self,
-                builder,
-                graphql_type_name,
-                model_name,
-            ),
             types::TypeId::ScalarTypeComparisonExpression {
                 scalar_type_name: _,
                 graphql_type_name,
@@ -224,6 +225,10 @@ pub enum Error {
         "internal error while building schema, filter_expression for model not found: {model_name}"
     )]
     InternalModelFilterExpressionNotFound { model_name: Qualified<ModelName> },
+    #[error("internal error while building schema, boolean expression not found: {type_name}")]
+    InternalBooleanExpressionNotFound {
+        type_name: Qualified<CustomTypeName>,
+    },
     #[error(
         "Conflicting argument names {argument_name} for field {field_name} of type {type_name}"
     )]

v3/crates/engine/src/schema/boolean_expression.rs

@@ -0,0 +1,275 @@
+use hasura_authn_core::Role;
+use lang_graphql::ast::common as ast;
+use lang_graphql::schema::{self as gql_schema};
+use open_dds::types::CustomTypeName;
+use std::collections::{BTreeMap, HashMap};
+
+use super::types::output_type::get_object_type_representation;
+use super::types::output_type::relationship::{FilterRelationshipAnnotation, ModelTargetSource};
+use super::types::{BooleanExpressionAnnotation, InputAnnotation, TypeId};
+use crate::metadata::resolved;
+use crate::metadata::resolved::relationship::{
+    relationship_execution_category, RelationshipExecutionCategory, RelationshipTarget,
+};
+use crate::metadata::resolved::subgraph::Qualified;
+use crate::metadata::resolved::types::mk_name;
+
+use crate::schema::permissions;
+use crate::schema::types;
+use crate::schema::GDS;
+
+type Error = crate::schema::Error;
+
+pub fn build_boolean_expression_input_schema(
+    gds: &GDS,
+    builder: &mut gql_schema::Builder<GDS>,
+    type_name: &ast::TypeName,
+    gds_type_name: &Qualified<CustomTypeName>,
+) -> Result<gql_schema::TypeInfo<GDS>, Error> {
+    let object_boolean_expression_type =
+        gds.metadata
+            .boolean_expression_types
+            .get(gds_type_name)
+            .ok_or_else(|| crate::schema::Error::InternalTypeNotFound {
+                type_name: gds_type_name.clone(),
+            })?;
+
+    if let Some(boolean_expression_info) = &object_boolean_expression_type.graphql {
+        let mut input_fields = BTreeMap::new();
+
+        // `_and`, `_or` or `_not` fields are available for all roles
+        let not_field_name = &boolean_expression_info.graphql_config.not_operator_name;
+
+        input_fields.insert(
+            not_field_name.clone(),
+            builder.allow_all_namespaced(
+                gql_schema::InputField::<GDS>::new(
+                    not_field_name.clone(),
+                    None,
+                    types::Annotation::Input(InputAnnotation::BooleanExpression(
+                        BooleanExpressionAnnotation::BooleanExpressionArgument {
+                            field: types::ModelFilterArgument::NotOp,
+                        },
+                    )),
+                    ast::TypeContainer::named_null(gql_schema::RegisteredTypeName::new(
+                        type_name.0.clone(),
+                    )),
+                    None,
+                    gql_schema::DeprecationStatus::NotDeprecated,
+                ),
+                None,
+            ),
+        );
+
+        let and_field_name = &boolean_expression_info.graphql_config.and_operator_name;
+
+        input_fields.insert(
+            and_field_name.clone(),
+            builder.allow_all_namespaced(
+                gql_schema::InputField::<GDS>::new(
+                    and_field_name.clone(),
+                    None,
+                    types::Annotation::Input(InputAnnotation::BooleanExpression(
+                        BooleanExpressionAnnotation::BooleanExpressionArgument {
+                            field: types::ModelFilterArgument::AndOp,
+                        },
+                    )),
+                    ast::TypeContainer::list_null(ast::TypeContainer::named_non_null(
+                        gql_schema::RegisteredTypeName::new(type_name.0.clone()),
+                    )),
+                    None,
+                    gql_schema::DeprecationStatus::NotDeprecated,
+                ),
+                None,
+            ),
+        );
+
+        let or_field_name = &boolean_expression_info.graphql_config.or_operator_name;
+        input_fields.insert(
+            or_field_name.clone(),
+            builder.allow_all_namespaced(
+                gql_schema::InputField::<GDS>::new(
+                    or_field_name.clone(),
+                    None,
+                    types::Annotation::Input(InputAnnotation::BooleanExpression(
+                        BooleanExpressionAnnotation::BooleanExpressionArgument {
+                            field: types::ModelFilterArgument::OrOp,
+                        },
+                    )),
+                    ast::TypeContainer::list_null(ast::TypeContainer::named_non_null(
+                        gql_schema::RegisteredTypeName::new(type_name.0.clone()),
+                    )),
+                    None,
+                    gql_schema::DeprecationStatus::NotDeprecated,
+                ),
+                None,
+            ),
+        );
+
+        let object_type_representation =
+            get_object_type_representation(gds, &object_boolean_expression_type.object_type)?;
+
+        // column fields
+        for (field_name, comparison_expression) in &boolean_expression_info.scalar_fields {
+            let field_graphql_name = mk_name(field_name.clone().0.as_str())?;
+            let registered_type_name =
+                get_scalar_comparison_input_type(builder, comparison_expression)?;
+            let field_type = ast::TypeContainer::named_null(registered_type_name);
+            let annotation = types::Annotation::Input(InputAnnotation::BooleanExpression(
+                BooleanExpressionAnnotation::BooleanExpressionArgument {
+                    field: types::ModelFilterArgument::Field {
+                        ndc_column: comparison_expression.ndc_column.clone(),
+                    },
+                },
+            ));
+            let field_permissions: HashMap<Role, Option<types::NamespaceAnnotation>> =
+                permissions::get_allowed_roles_for_field(object_type_representation, field_name)
+                    .map(|role| (role.clone(), None))
+                    .collect();
+
+            let input_field = builder.conditional_namespaced(
+                gql_schema::InputField::<GDS>::new(
+                    field_graphql_name.clone(),
+                    None,
+                    annotation,
+                    field_type,
+                    None,
+                    gql_schema::DeprecationStatus::NotDeprecated,
+                ),
+                field_permissions,
+            );
+            input_fields.insert(field_graphql_name, input_field);
+        }
+
+        // relationship fields
+        // TODO(naveen): Add support for command relationships
+        for (rel_name, relationship) in object_type_representation.relationships.iter() {
+            if let RelationshipTarget::Model {
+                model_name,
+                relationship_type,
+                target_typename,
+                mappings,
+            } = &relationship.target
+            {
+                let target_model = gds.metadata.models.get(model_name).ok_or_else(|| {
+                    crate::schema::Error::InternalModelNotFound {
+                        model_name: model_name.clone(),
+                    }
+                })?;
+
+                let target_object_type_representation =
+                    get_object_type_representation(gds, &target_model.data_type)?;
+
+                // Build relationship field in filter expression only when
+                // the target_model is backed by a source
+                if let Some(target_source) = &target_model.source {
+                    let target_model_source =
+                        ModelTargetSource::from_model_source(target_source, relationship)?;
+
+                    // filter expression with relationships is currently only supported for local relationships
+                    if let RelationshipExecutionCategory::Local = relationship_execution_category(
+                        &object_boolean_expression_type.data_connector_link,
+                        &target_source.data_connector,
+                        &target_model_source.capabilities,
+                    ) {
+                        if target_source.data_connector.name
+                            == object_boolean_expression_type.data_connector_name
+                        {
+                            // If the relationship target model does not have filterExpressionType do not include
+                            // it in the source model filter expression input type.
+                            if let Some(ref target_model_filter_expression) = &target_model
+                                .clone()
+                                .filter_expression_type
+                                .and_then(|ref object_boolean_expression_type| {
+                                    object_boolean_expression_type.clone().graphql
+                                })
+                            {
+                                let target_model_filter_expression_type_name =
+                                    &target_model_filter_expression.type_name;
+
+                                let annotation = FilterRelationshipAnnotation {
+                                    source_type: relationship.source.clone(),
+                                    relationship_name: relationship.name.clone(),
+                                    target_source: target_model_source.clone(),
+                                    target_type: target_typename.clone(),
+                                    target_model_name: target_model.name.clone(),
+                                    relationship_type: relationship_type.clone(),
+                                    mappings: mappings.clone(),
+                                    source_data_connector: object_boolean_expression_type
+                                        .data_connector_link
+                                        .clone(),
+                                    source_type_mappings: object_boolean_expression_type
+                                        .type_mappings
+                                        .clone(),
+                                };
+
+                                let namespace_annotations =
+                                    permissions::get_model_relationship_namespace_annotations(
+                                        target_model,
+                                        object_type_representation,
+                                        target_object_type_representation,
+                                        mappings,
+                                        &gds.metadata.object_types,
+                                    )?;
+
+                                input_fields.insert(
+                                    rel_name.clone(),
+                                    builder.conditional_namespaced(
+                                        gql_schema::InputField::<GDS>::new(
+                                            rel_name.clone(),
+                                            None,
+                                            types::Annotation::Input(InputAnnotation::BooleanExpression(
+                                                BooleanExpressionAnnotation
+                                                ::BooleanExpressionArgument {
+                                                    field:
+                                                        types::ModelFilterArgument::RelationshipField(
+                                                            annotation,
+                                                        ),
+                                                },
+                                            )),
+                                            ast::TypeContainer::named_null(
+                                                gql_schema::RegisteredTypeName::new(
+                                                    target_model_filter_expression_type_name.0.clone(),
+                                                ),
+                                            ),
+                                            None,
+                                            gql_schema::DeprecationStatus::NotDeprecated,
+                                        ),
+                                        namespace_annotations
+                                    ),
+                                );
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        Ok(gql_schema::TypeInfo::InputObject(
+            gql_schema::InputObject::new(type_name.clone(), None, input_fields, Vec::new()),
+        ))
+    } else {
+        Err(crate::schema::Error::InternalBooleanExpressionNotFound {
+            type_name: gds_type_name.clone(),
+        })
+    }
+}
+
+fn get_scalar_comparison_input_type(
+    builder: &mut gql_schema::Builder<GDS>,
+    comparison_expression: &resolved::boolean_expression::ComparisonExpressionInfo,
+) -> Result<gql_schema::RegisteredTypeName, Error> {
+    let graphql_type_name = comparison_expression.type_name.clone();
+    let mut operators = Vec::new();
+    for (op_name, input_type) in &comparison_expression.operators {
+        let op_name = mk_name(op_name.as_str())?;
+        operators.push((op_name, input_type.clone()))
+    }
+    Ok(
+        builder.register_type(TypeId::ScalarTypeComparisonExpression {
+            scalar_type_name: comparison_expression.scalar_type_name.clone(),
+            graphql_type_name,
+            operators,
+            is_null_operator_name: mk_name(&comparison_expression.is_null_operator_name)?,
+        }),
+    )
+}

v3/crates/engine/src/schema/model_filter.rs

@@ -1,20 +1,13 @@
-use hasura_authn_core::Role;
 use lang_graphql::ast::common as ast;
 use lang_graphql::schema::{self as gql_schema, InputField, Namespaced};
-use open_dds::models::ModelName;
+use open_dds::types::CustomTypeName;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
 
-use super::types::output_type::get_object_type_representation;
+use super::types::input_type;
-use super::types::output_type::relationship::{FilterRelationshipAnnotation, ModelTargetSource};
-use super::types::{input_type, InputAnnotation, ModelInputAnnotation, TypeId};
 use crate::metadata::resolved;
-use crate::metadata::resolved::model::ComparisonExpressionInfo;
+
-use crate::metadata::resolved::relationship::{
-    relationship_execution_category, RelationshipExecutionCategory, RelationshipTarget,
-};
 use crate::metadata::resolved::subgraph::{Qualified, QualifiedTypeReference};
-use crate::metadata::resolved::types::mk_name;
+
-use crate::schema::permissions;
 use crate::schema::types;
 use crate::schema::GDS;
 
@@ -22,22 +15,22 @@ type Error = crate::schema::Error;
 
 pub fn get_where_expression_input_field(
     builder: &mut gql_schema::Builder<GDS>,
-    model_name: Qualified<ModelName>,
+    gds_type_name: Qualified<CustomTypeName>,
-    boolean_expression_info: &resolved::model::ModelFilterExpression,
+    boolean_expression_info: &resolved::boolean_expression::BooleanExpression,
 ) -> gql_schema::InputField<GDS> {
     gql_schema::InputField::new(
         boolean_expression_info
-            .filter_graphql_config
+            .graphql_config
             .where_field_name
             .clone(),
         None,
-        types::Annotation::Input(types::InputAnnotation::Model(
+        types::Annotation::Input(types::InputAnnotation::BooleanExpression(
-            types::ModelInputAnnotation::ModelFilterExpression,
+            types::BooleanExpressionAnnotation::BooleanExpression,
         )),
         ast::TypeContainer::named_null(builder.register_type(
-            types::TypeId::ModelBooleanExpression {
+            types::TypeId::InputObjectBooleanExpressionType {
-                model_name,
+                gds_type_name,
-                graphql_type_name: boolean_expression_info.where_type_name.clone(),
+                graphql_type_name: boolean_expression_info.type_name.clone(),
             },
         )),
         None,
@@ -45,257 +38,6 @@ pub fn get_where_expression_input_field(
     )
 }
 
-pub fn build_model_filter_expression_input_schema(
-    gds: &GDS,
-    builder: &mut gql_schema::Builder<GDS>,
-    type_name: &ast::TypeName,
-    model_name: &Qualified<ModelName>,
-) -> Result<gql_schema::TypeInfo<GDS>, Error> {
-    let model = gds.metadata.models.get(model_name).ok_or_else(|| {
-        crate::schema::Error::InternalModelNotFound {
-            model_name: model_name.clone(),
-        }
-    })?;
-    if let Some(boolean_expression_info) = &model.graphql_api.filter_expression {
-        let mut input_fields = BTreeMap::new();
-
-        // `_and`, `_or` or `_not` fields are available for all roles
-        let not_field_name = &boolean_expression_info
-            .filter_graphql_config
-            .not_operator_name;
-
-        input_fields.insert(
-            not_field_name.clone(),
-            builder.allow_all_namespaced(
-                gql_schema::InputField::<GDS>::new(
-                    not_field_name.clone(),
-                    None,
-                    types::Annotation::Input(InputAnnotation::Model(
-                        ModelInputAnnotation::ModelFilterArgument {
-                            field: types::ModelFilterArgument::NotOp,
-                        },
-                    )),
-                    ast::TypeContainer::named_null(gql_schema::RegisteredTypeName::new(
-                        type_name.0.clone(),
-                    )),
-                    None,
-                    gql_schema::DeprecationStatus::NotDeprecated,
-                ),
-                None,
-            ),
-        );
-
-        let and_field_name = &boolean_expression_info
-            .filter_graphql_config
-            .and_operator_name;
-
-        input_fields.insert(
-            and_field_name.clone(),
-            builder.allow_all_namespaced(
-                gql_schema::InputField::<GDS>::new(
-                    and_field_name.clone(),
-                    None,
-                    types::Annotation::Input(InputAnnotation::Model(
-                        ModelInputAnnotation::ModelFilterArgument {
-                            field: types::ModelFilterArgument::AndOp,
-                        },
-                    )),
-                    ast::TypeContainer::list_null(ast::TypeContainer::named_non_null(
-                        gql_schema::RegisteredTypeName::new(type_name.0.clone()),
-                    )),
-                    None,
-                    gql_schema::DeprecationStatus::NotDeprecated,
-                ),
-                None,
-            ),
-        );
-
-        let or_field_name = &boolean_expression_info
-            .filter_graphql_config
-            .or_operator_name;
-        input_fields.insert(
-            or_field_name.clone(),
-            builder.allow_all_namespaced(
-                gql_schema::InputField::<GDS>::new(
-                    or_field_name.clone(),
-                    None,
-                    types::Annotation::Input(InputAnnotation::Model(
-                        ModelInputAnnotation::ModelFilterArgument {
-                            field: types::ModelFilterArgument::OrOp,
-                        },
-                    )),
-                    ast::TypeContainer::list_null(ast::TypeContainer::named_non_null(
-                        gql_schema::RegisteredTypeName::new(type_name.0.clone()),
-                    )),
-                    None,
-                    gql_schema::DeprecationStatus::NotDeprecated,
-                ),
-                None,
-            ),
-        );
-
-        let object_type_representation = get_object_type_representation(gds, &model.data_type)?;
-
-        // column fields
-        if let Some(model_filter_expression) = model.graphql_api.filter_expression.as_ref() {
-            for (field_name, comparison_expression) in &model_filter_expression.scalar_fields {
-                let field_graphql_name = mk_name(field_name.clone().0.as_str())?;
-                let registered_type_name =
-                    get_scalar_comparison_input_type(builder, comparison_expression)?;
-                let field_type = ast::TypeContainer::named_null(registered_type_name);
-                let annotation = types::Annotation::Input(InputAnnotation::Model(
-                    ModelInputAnnotation::ModelFilterArgument {
-                        field: types::ModelFilterArgument::Field {
-                            ndc_column: comparison_expression.ndc_column.clone(),
-                        },
-                    },
-                ));
-                let field_permissions: HashMap<Role, Option<types::NamespaceAnnotation>> =
-                    permissions::get_allowed_roles_for_field(
-                        object_type_representation,
-                        field_name,
-                    )
-                    .map(|role| (role.clone(), None))
-                    .collect();
-
-                let input_field = builder.conditional_namespaced(
-                    gql_schema::InputField::<GDS>::new(
-                        field_graphql_name.clone(),
-                        None,
-                        annotation,
-                        field_type,
-                        None,
-                        gql_schema::DeprecationStatus::NotDeprecated,
-                    ),
-                    field_permissions,
-                );
-                input_fields.insert(field_graphql_name, input_field);
-            }
-        }
-
-        // relationship fields
-        // TODO(naveen): Add support for command relationships
-        for (rel_name, relationship) in object_type_representation.relationships.iter() {
-            if let RelationshipTarget::Model {
-                model_name,
-                relationship_type,
-                target_typename,
-                mappings,
-            } = &relationship.target
-            {
-                let target_model = gds.metadata.models.get(model_name).ok_or_else(|| {
-                    crate::schema::Error::InternalModelNotFound {
-                        model_name: model_name.clone(),
-                    }
-                })?;
-
-                let target_object_type_representation =
-                    get_object_type_representation(gds, &target_model.data_type)?;
-
-                // Build relationship field in filter expression only when both
-                // the target_model and source model are backed by a source
-                if let (Some(target_source), Some(model_source)) =
-                    (&target_model.source, &model.source)
-                {
-                    let target_model_source =
-                        ModelTargetSource::from_model_source(target_source, relationship)?;
-
-                    // filter expression with relationships is currently only supported for local relationships
-                    if let RelationshipExecutionCategory::Local = relationship_execution_category(
-                        &model_source.data_connector,
-                        &target_source.data_connector,
-                        &target_model_source.capabilities,
-                    ) {
-                        if target_source.data_connector.name == model_source.data_connector.name {
-                            // If the relationship target model does not have filterExpressionType do not include
-                            // it in the source model filter expression input type.
-                            if let Some(target_model_filter_expression) =
-                                target_model.graphql_api.filter_expression.as_ref()
-                            {
-                                let target_model_filter_expression_type_name =
-                                    &target_model_filter_expression.where_type_name;
-
-                                let annotation = FilterRelationshipAnnotation {
-                                    source_type: relationship.source.clone(),
-                                    relationship_name: relationship.name.clone(),
-                                    target_source: target_model_source.clone(),
-                                    target_type: target_typename.clone(),
-                                    target_model_name: target_model.name.clone(),
-                                    relationship_type: relationship_type.clone(),
-                                    mappings: mappings.clone(),
-                                    source_data_connector: model_source.data_connector.clone(),
-                                    source_type_mappings: model_source.type_mappings.clone(),
-                                };
-
-                                input_fields.insert(
-                                    rel_name.clone(),
-                                    builder.conditional_namespaced(
-                                        gql_schema::InputField::<GDS>::new(
-                                            rel_name.clone(),
-                                            None,
-                                            types::Annotation::Input(InputAnnotation::Model(
-                                                ModelInputAnnotation::ModelFilterArgument {
-                                                    field:
-                                                        types::ModelFilterArgument::RelationshipField(
-                                                            annotation,
-                                                        ),
-                                                },
-                                            )),
-                                            ast::TypeContainer::named_null(
-                                                gql_schema::RegisteredTypeName::new(
-                                                    target_model_filter_expression_type_name.0.clone(),
-                                                ),
-                                            ),
-                                            None,
-                                            gql_schema::DeprecationStatus::NotDeprecated,
-                                        ),
-                                        permissions::get_model_relationship_namespace_annotations(
-                                            target_model,
-                                            object_type_representation,
-                                            target_object_type_representation,
-                                            mappings,
-                                            &gds.metadata.object_types,
-                                        )?,
-                                    ),
-                                );
-                            }
-                        }
-                    }
-                }
-            }
-        }
-        Ok(gql_schema::TypeInfo::InputObject(
-            gql_schema::InputObject::new(type_name.clone(), None, input_fields, Vec::new()),
-        ))
-    } else {
-        Err(
-            crate::schema::Error::InternalModelFilterExpressionNotFound {
-                model_name: model_name.clone(),
-            },
-        )
-    }
-}
-
-fn get_scalar_comparison_input_type(
-    builder: &mut gql_schema::Builder<GDS>,
-    comparison_expression: &ComparisonExpressionInfo,
-) -> Result<gql_schema::RegisteredTypeName, Error> {
-    let graphql_type_name = comparison_expression.type_name.clone();
-    let mut operators = Vec::new();
-    for (op_name, input_type) in &comparison_expression.operators {
-        let op_name = mk_name(op_name.as_str())?;
-        operators.push((op_name, input_type.clone()))
-    }
-    Ok(
-        builder.register_type(TypeId::ScalarTypeComparisonExpression {
-            scalar_type_name: comparison_expression.scalar_type_name.clone(),
-            graphql_type_name,
-            operators,
-            is_null_operator_name: mk_name(&comparison_expression.is_null_operator_name)?,
-        }),
-    )
-}
-
 pub fn build_scalar_comparison_input(
     gds: &GDS,
     builder: &mut gql_schema::Builder<GDS>,

v3/crates/engine/src/schema/query_root/select_many.rs

@@ -75,14 +75,20 @@ pub(crate) fn generate_select_many_arguments(
     }
 
     // generate and insert where argument
-    if let Some(filter_expression_info) = &model.graphql_api.filter_expression {
+    if let Some(object_boolean_expression_type) = &model.filter_expression_type {
-        let where_argument =
+        if let Some(boolean_expression) = &object_boolean_expression_type.graphql {
-            get_where_expression_input_field(builder, model.name.clone(), filter_expression_info);
+            let where_argument = get_where_expression_input_field(
+                builder,
+                object_boolean_expression_type.name.clone(),
+                boolean_expression,
+            );
+
             arguments.insert(
                 where_argument.name.clone(),
                 builder.allow_all_namespaced(where_argument, None),
             );
         }
+    }
 
     Ok(arguments)
 }

v3/crates/engine/src/schema/types.rs

@@ -200,10 +200,6 @@ pub enum ModelInputAnnotation {
         argument_type: QualifiedTypeReference,
         ndc_table_argument: Option<String>,
     },
-    ModelFilterExpression,
-    ModelFilterArgument {
-        field: ModelFilterArgument,
-    },
     ComparisonOperation {
         operator: String,
     },
@@ -225,6 +221,13 @@ pub enum ModelInputAnnotation {
     },
 }
 
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Display)]
+/// Annotations of the input types/fields related to boolean expressions.
+pub enum BooleanExpressionAnnotation {
+    BooleanExpression,
+    BooleanExpressionArgument { field: ModelFilterArgument },
+}
+
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Display)]
 /// Annotations for Relay input arguments/types.
 pub enum RelayInputAnnotation {
@@ -245,6 +248,7 @@ pub enum InputAnnotation {
         field_name: types::FieldName,
         field_type: QualifiedTypeReference,
     },
+    BooleanExpression(BooleanExpressionAnnotation),
     CommandArgument {
         argument_type: QualifiedTypeReference,
         ndc_func_proc_argument: Option<String>,
@@ -341,15 +345,15 @@ pub enum TypeId {
         gds_type_name: Qualified<types::CustomTypeName>,
         graphql_type_name: ast::TypeName,
     },
+    InputObjectBooleanExpressionType {
+        gds_type_name: Qualified<types::CustomTypeName>,
+        graphql_type_name: ast::TypeName,
+    },
     NodeRoot,
     ModelArgumentsInput {
         model_name: Qualified<models::ModelName>,
         type_name: ast::TypeName,
     },
-    ModelBooleanExpression {
-        model_name: Qualified<models::ModelName>,
-        graphql_type_name: ast::TypeName,
-    },
     ModelOrderByExpression {
         model_name: Qualified<models::ModelName>,
         graphql_type_name: ast::TypeName,
@@ -396,7 +400,7 @@ impl TypeId {
             } => graphql_type_name.clone(),
             TypeId::NodeRoot => ast::TypeName(mk_name!("Node")),
             TypeId::ModelArgumentsInput { type_name, .. } => type_name.clone(),
-            TypeId::ModelBooleanExpression {
+            TypeId::InputObjectBooleanExpressionType {
                 graphql_type_name, ..
             } => graphql_type_name.clone(),
             TypeId::ScalarTypeComparisonExpression {

v3/crates/engine/tests/execute/models/select_many/where/shared_boolean_expression/expected.json

@@ -0,0 +1,138 @@
+[
+  {
+    "data": {
+      "AuthorManyNot": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorManyOr": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorManyAnd": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorManyNestedBool": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorTwoManyNot": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorTwoManyOr": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorTwoManyAnd": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorTwoManyNestedBool": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ]
+    }
+  },
+  {
+    "data": {
+      "AuthorManyNot": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorManyOr": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorManyAnd": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorManyNestedBool": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorTwoManyNot": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ],
+      "AuthorTwoManyOr": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorTwoManyAnd": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        }
+      ],
+      "AuthorTwoManyNestedBool": [
+        {
+          "author_id": 1,
+          "first_name": "Peter"
+        },
+        {
+          "author_id": 2,
+          "first_name": "John"
+        }
+      ]
+    }
+  }
+]

v3/crates/engine/tests/execute/models/select_many/where/shared_boolean_expression/metadata.json

@@ -0,0 +1,263 @@
+{
+  "version": "v2",
+  "subgraphs": [
+    {
+      "name": "default",
+      "objects": [
+        {
+          "kind": "DataConnectorScalarRepresentation",
+          "version": "v1",
+          "definition": {
+            "dataConnectorName": "db",
+            "dataConnectorScalarType": "String",
+            "representation": "String",
+            "graphql": {
+              "comparisonExpressionTypeName": "String_Comparison_Exp"
+            }
+          }
+        },
+        {
+          "definition": {
+            "dataConnectorName": "db",
+            "dataConnectorScalarType": "int4",
+            "representation": "Int",
+            "graphql": {
+              "comparisonExpressionTypeName": "int4_comparison"
+            }
+          },
+          "version": "v1",
+          "kind": "DataConnectorScalarRepresentation"
+        },
+        {
+          "kind": "ObjectType",
+          "version": "v1",
+          "definition": {
+            "name": "author",
+            "fields": [
+              {
+                "name": "author_id",
+                "type": "Int!"
+              },
+              {
+                "name": "first_name",
+                "type": "String!"
+              },
+              {
+                "name": "last_name",
+                "type": "String!"
+              }
+            ],
+            "graphql": {
+              "typeName": "Author"
+            },
+            "dataConnectorTypeMapping": [
+              {
+                "dataConnectorName": "db",
+                "dataConnectorObjectType": "author",
+                "fieldMapping": {
+                  "author_id": {
+                    "column": {
+                      "name": "id"
+                    }
+                  },
+                  "first_name": {
+                    "column": {
+                      "name": "first_name"
+                    }
+                  },
+                  "last_name": {
+                    "column": {
+                      "name": "last_name"
+                    }
+                  }
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "ObjectBooleanExpressionType",
+          "version": "v1",
+          "definition": {
+            "name": "author_bool_exp",
+            "objectType": "author",
+            "dataConnectorName": "db",
+            "dataConnectorObjectType": "author",
+            "comparableFields": [
+              {
+                "fieldName": "author_id",
+                "operators": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "first_name",
+                "operators": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "last_name",
+                "operators": {
+                  "enableAll": true
+                }
+              }
+            ],
+            "graphql": {
+              "typeName": "AuthorFilter"
+            }
+          }
+        },
+        {
+          "kind": "Model",
+          "version": "v1",
+          "definition": {
+            "name": "AuthorsTwo",
+            "objectType": "author",
+            "source": {
+              "dataConnectorName": "db",
+              "collection": "author"
+            },
+            "graphql": {
+              "selectUniques": [],
+              "selectMany": {
+                "queryRootField": "AuthorTwoMany"
+              }
+            },
+            "filterExpressionType": "author_bool_exp",
+            "orderableFields": [
+              {
+                "fieldName": "author_id",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "first_name",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "last_name",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "Model",
+          "version": "v1",
+          "definition": {
+            "name": "Authors",
+            "objectType": "author",
+            "source": {
+              "dataConnectorName": "db",
+              "collection": "author"
+            },
+            "graphql": {
+              "selectUniques": [],
+              "selectMany": {
+                "queryRootField": "AuthorMany"
+              }
+            },
+            "filterExpressionType": "author_bool_exp",
+            "orderableFields": [
+              {
+                "fieldName": "author_id",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "first_name",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "last_name",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "TypePermissions",
+          "version": "v1",
+          "definition": {
+            "typeName": "author",
+            "permissions": [
+              {
+                "role": "admin",
+                "output": {
+                  "allowedFields": [
+                    "author_id",
+                    "first_name",
+                    "last_name"
+                  ]
+                }
+              },
+              {
+                "role": "user",
+                "output": {
+                  "allowedFields": [
+                    "author_id",
+                    "first_name",
+                    "last_name"
+                  ]
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "ModelPermissions",
+          "version": "v1",
+          "definition": {
+            "modelName": "Authors",
+            "permissions": [
+              {
+                "role": "admin",
+                "select": {
+                  "filter": null
+                }
+              },
+              {
+                "role": "user",
+                "select": {
+                  "filter": null
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "ModelPermissions",
+          "version": "v1",
+          "definition": {
+            "modelName": "AuthorsTwo",
+            "permissions": [
+              {
+                "role": "admin",
+                "select": {
+                  "filter": null
+                }
+              },
+              {
+                "role": "user",
+                "select": {
+                  "filter": null
+                }
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}

v3/crates/engine/tests/execute/models/select_many/where/shared_boolean_expression/request.gql

@@ -0,0 +1,28 @@
+query {
+    AuthorManyNot: AuthorMany(where: {_not: {first_name: {_is_null: true}}}) {
+      author_id first_name
+    }
+    AuthorManyOr: AuthorMany(where: {_or: [{first_name: {_like: "Pet%"}}]}) {
+      author_id first_name
+    }
+    AuthorManyAnd: AuthorMany(where: {_and: [{first_name: {_like: "Pet%"}}]}) {
+      author_id first_name
+    }
+    AuthorManyNestedBool: AuthorMany(where: {_or: [{_and: [{author_id: {_eq: 1}}, {first_name: {_eq: "Peter"}}]}, {_and: [{author_id: {_eq: 2}}, {first_name: {_eq: "John"}}]}]}){
+      author_id
+      first_name
+    }
+    AuthorTwoManyNot: AuthorTwoMany(where: {_not: {first_name: {_is_null: true}}}) {
+      author_id first_name
+    }
+    AuthorTwoManyOr: AuthorTwoMany(where: {_or: [{first_name: {_like: "Pet%"}}]}) {
+      author_id first_name
+    }
+    AuthorTwoManyAnd: AuthorTwoMany(where: {_and: [{first_name: {_like: "Pet%"}}]}) {
+      author_id first_name
+    }
+    AuthorTwoManyNestedBool: AuthorTwoMany(where: {_or: [{_and: [{author_id: {_eq: 1}}, {first_name: {_eq: "Peter"}}]}, {_and: [{author_id: {_eq: 2}}, {first_name: {_eq: "John"}}]}]}){
+      author_id
+      first_name
+    }
+}

v3/crates/engine/tests/execute/models/select_many/where/shared_boolean_expression/session_variables.json

@@ -0,0 +1,9 @@
+[
+    {
+        "x-hasura-role": "admin"
+    },
+    {
+        "x-hasura-role": "user",
+        "x-hasura-user-id": "2"
+    }
+]

v3/crates/engine/tests/execution.rs

@@ -206,6 +206,15 @@ fn test_model_select_many_where() -> anyhow::Result<()> {
     common::test_execution_expectation(test_path_string, &[common_metadata_path_string])
 }
 
+// the test here is that two Models can both use the same ObjectBooleanExpressionType without
+// errors
+#[test]
+fn test_model_select_many_shared_boolean_expression() -> anyhow::Result<()> {
+    let test_path_string = "execute/models/select_many/where/shared_boolean_expression";
+    let common_metadata_path_string = "execute/common_metadata/postgres_connector_schema.json";
+    common::test_execution_expectation(test_path_string, &[common_metadata_path_string])
+}
+
 #[test]
 fn test_model_select_many_where_is_null() -> anyhow::Result<()> {
     let test_path_string = "execute/models/select_many/where/is_null";

v3/crates/lang-graphql/src/validation/input/normalize.rs

@@ -33,6 +33,7 @@ pub fn normalize<'q, 's, S: schema::SchemaContext, V: ValueSource<'q, 's, S>>(
 ) -> Result<normalized::Value<'s, S>>
 where
     's: 'q,
+    V: std::fmt::Debug,
 {
     let normalized_value = match &location_type.type_().base {
         ast::BaseType::Named(_) => {
@@ -191,6 +192,7 @@ fn normalize_input_object<'q, 's, S: schema::SchemaContext, V: ValueSource<'q, '
 ) -> Result<normalized::Value<'s, S>>
 where
     's: 'q,
+    V: std::fmt::Debug,
 {
     // let (normalized_object, required_field_count) = value.fold_key_values(
     let normalized_object = value.fold_key_values(