Separate relationships resolve stage (#522)

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

## Description

This separates out the stage that resolves relationships. The most
important thing here is that we no longer have a `relationships` field
in `ObjectTypeRepresentation` that may or may not be populated, and
instead add a new wrapper type `ObjectTypeWithRelationships`, which is
used downstream of this stage.

Stacked on top of https://github.com/hasura/v3-engine/pull/521

Functional no-op.

V3_GIT_ORIGIN_REV_ID: 1e6ca41e55b8cc470385c35bbd7999fa7a2bce6e

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

@@ -23,10 +23,7 @@ use super::{
 };
 
 use crate::execute::model_tracking::{count_model, UsagesCounts};
-use crate::metadata::resolved::{
+use crate::metadata::resolved::subgraph::serialize_qualified_btreemap;
-    relationship::{relationship_execution_category, RelationshipExecutionCategory},
-    subgraph::serialize_qualified_btreemap,
-};
 use crate::schema::types::output_type::relationship::{
     ModelRelationshipAnnotation, ModelTargetSource,
 };
@@ -54,7 +51,7 @@ pub(crate) struct LocalModelRelationshipInfo<'s> {
     pub source_type_mappings: &'s BTreeMap<Qualified<CustomTypeName>, resolved::TypeMapping>,
     pub target_source: &'s ModelTargetSource,
     pub target_type: &'s Qualified<CustomTypeName>,
-    pub mappings: &'s Vec<resolved::relationship::RelationshipModelMapping>,
+    pub mappings: &'s Vec<resolved::RelationshipModelMapping>,
 }
 
 #[derive(Debug, Serialize)]
@@ -100,19 +97,19 @@ pub(crate) fn process_model_relationship_definition(
     } = relationship_info;
 
     let mut column_mapping = BTreeMap::new();
-    for resolved::relationship::RelationshipModelMapping {
+    for resolved::RelationshipModelMapping {
         source_field: source_field_path,
         target_field: _,
         target_ndc_column,
     } in mappings.iter()
     {
         if !matches!(
-            relationship_execution_category(
+            resolved::relationship_execution_category(
                 source_data_connector,
                 &target_source.model.data_connector,
                 &target_source.capabilities
             ),
-            RelationshipExecutionCategory::Local
+            resolved::RelationshipExecutionCategory::Local
         ) {
             Err(error::InternalEngineError::RemoteRelationshipsAreNotSupported)?
         } else {
@@ -165,18 +162,18 @@ pub(crate) fn process_command_relationship_definition(
     } = relationship_info;
 
     let mut arguments = BTreeMap::new();
-    for resolved::relationship::RelationshipCommandMapping {
+    for resolved::RelationshipCommandMapping {
         source_field: source_field_path,
         argument_name: target_argument,
     } in annotation.mappings.iter()
     {
         if !matches!(
-            relationship_execution_category(
+            resolved::relationship_execution_category(
                 source_data_connector,
                 &target_source.details.data_connector,
                 &target_source.capabilities
             ),
-            RelationshipExecutionCategory::Local
+            resolved::RelationshipExecutionCategory::Local
         ) {
             Err(error::InternalEngineError::RemoteRelationshipsAreNotSupported)?
         } else {
@@ -293,12 +290,12 @@ pub(crate) fn generate_model_relationship_ir<'s>(
                 }
                 None => error::Error::from(normalized_ast::Error::NoTypenameFound),
             })?;
-    match relationship_execution_category(
+    match resolved::relationship_execution_category(
         source_data_connector,
         &target_source.model.data_connector,
         &target_source.capabilities,
     ) {
-        RelationshipExecutionCategory::Local => build_local_model_relationship(
+        resolved::RelationshipExecutionCategory::Local => build_local_model_relationship(
             field,
             field_call,
             annotation,
@@ -312,7 +309,7 @@ pub(crate) fn generate_model_relationship_ir<'s>(
             session_variables,
             usage_counts,
         ),
-        RelationshipExecutionCategory::RemoteForEach => build_remote_relationship(
+        resolved::RelationshipExecutionCategory::RemoteForEach => build_remote_relationship(
             field,
             field_call,
             annotation,
@@ -353,12 +350,12 @@ pub(crate) fn generate_command_relationship_ir<'s>(
                 None => error::Error::from(normalized_ast::Error::NoTypenameFound),
             })?;
 
-    match relationship_execution_category(
+    match resolved::relationship_execution_category(
         source_data_connector,
         &target_source.details.data_connector,
         &target_source.capabilities,
     ) {
-        RelationshipExecutionCategory::Local => build_local_command_relationship(
+        resolved::RelationshipExecutionCategory::Local => build_local_command_relationship(
             field,
             field_call,
             annotation,
@@ -367,14 +364,16 @@ pub(crate) fn generate_command_relationship_ir<'s>(
             target_source,
             session_variables,
         ),
-        RelationshipExecutionCategory::RemoteForEach => build_remote_command_relationship(
+        resolved::RelationshipExecutionCategory::RemoteForEach => {
+            build_remote_command_relationship(
                 field,
                 field_call,
                 annotation,
                 type_mappings,
                 target_source,
                 session_variables,
-        ),
+            )
+        }
     }
 }
 
@@ -481,7 +480,7 @@ pub(crate) fn build_remote_relationship<'n, 's>(
     usage_counts: &mut UsagesCounts,
 ) -> Result<FieldSelection<'s>, error::Error> {
     let mut join_mapping: Vec<(SourceField, TargetField)> = vec![];
-    for resolved::relationship::RelationshipModelMapping {
+    for resolved::RelationshipModelMapping {
         source_field: source_field_path,
         target_field: target_field_path,
         target_ndc_column,
@@ -561,7 +560,7 @@ pub(crate) fn build_remote_command_relationship<'n, 's>(
     session_variables: &SessionVariables,
 ) -> Result<FieldSelection<'s>, error::Error> {
     let mut join_mapping: Vec<(SourceField, ArgumentName)> = vec![];
-    for resolved::relationship::RelationshipCommandMapping {
+    for resolved::RelationshipCommandMapping {
         source_field: source_field_path,
         argument_name: target_argument_name,
     } in annotation.mappings.iter()

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

@@ -5,7 +5,6 @@ pub mod metadata;
 pub mod model;
 pub mod ndc_validation;
 pub mod permission;
-pub mod relationship;
 pub mod stages;
 pub mod subgraph;
 mod typecheck;
@@ -23,4 +22,11 @@ pub use stages::models::{
     FilterPermission, Model, ModelOrderByExpression, ModelPredicate, ModelSource,
     SelectManyGraphQlDefinition, SelectUniqueGraphQlDefinition,
 };
+/// we seem to be exporting functions. perhaps these would be better served as methods on the data
+/// types we export?
+pub use stages::relationships::{
+    relationship_execution_category, ObjectTypeWithRelationships, Relationship,
+    RelationshipCapabilities, RelationshipCommandMapping, RelationshipExecutionCategory,
+    RelationshipModelMapping, RelationshipTarget,
+};
 pub use stages::resolve;

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

@@ -6,7 +6,7 @@ use crate::metadata::resolved::ndc_validation;
 use crate::metadata::resolved::permission::ValueExpression;
 use crate::metadata::resolved::stages::{
     boolean_expressions, data_connector_scalar_types, data_connector_type_mappings, models,
-    scalar_types, type_permissions,
+    relationships, scalar_types, type_permissions,
 };
 use crate::metadata::resolved::subgraph::{ArgumentInfo, Qualified};
 use crate::metadata::resolved::subgraph::{QualifiedBaseType, QualifiedTypeReference};
@@ -165,7 +165,7 @@ pub(crate) fn resolve_value_expression_for_argument(
     value_expression: &open_dds::permissions::ValueExpression,
     argument_type: &QualifiedTypeReference,
     subgraph: &str,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,
         boolean_expressions::ObjectBooleanExpressionType,

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

@@ -1,6 +1,6 @@
 use super::stages::{
     boolean_expressions, commands, data_connector_scalar_types, data_connector_type_mappings,
-    type_permissions,
+    relationships,
 };
 use crate::metadata::resolved::argument::resolve_value_expression_for_argument;
 use crate::metadata::resolved::error::Error;
@@ -17,7 +17,7 @@ use super::typecheck;
 pub fn resolve_command_permissions(
     command: &commands::Command,
     permissions: &CommandPermissionsV1,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,
         boolean_expressions::ObjectBooleanExpressionType,

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

@@ -9,19 +9,18 @@ use open_dds::{commands::CommandName, models::ModelName, types::CustomTypeName};
 use crate::metadata::resolved::command;
 use crate::metadata::resolved::error::Error;
 use crate::metadata::resolved::model::resolve_model_select_permissions;
-use crate::metadata::resolved::relationship::resolve_relationship;
 use crate::metadata::resolved::subgraph::Qualified;
 
 use crate::metadata::resolved::stages::{
     boolean_expressions, commands, data_connector_scalar_types, data_connector_type_mappings,
-    graphql_config, models, roles, scalar_types, type_permissions,
+    graphql_config, models, relationships, roles, scalar_types,
 };
 
 /// Resolved and validated metadata for a project. Used internally in the v3 server.
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
 pub struct Metadata {
     pub object_types:
-        HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+        HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     pub scalar_types: HashMap<Qualified<CustomTypeName>, scalar_types::ScalarTypeRepresentation>,
     pub models: IndexMap<Qualified<ModelName>, models::Model>,
     pub commands: IndexMap<Qualified<CommandName>, commands::Command>,
@@ -38,7 +37,7 @@ pub fn resolve_metadata(
     metadata_accessor: &open_dds::accessor::MetadataAccessor,
     graphql_config: &graphql_config::GraphqlConfig,
     data_connector_type_mappings: &data_connector_type_mappings::DataConnectorTypeMappings,
-    object_types: HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     scalar_types: &HashMap<Qualified<CustomTypeName>, scalar_types::ScalarTypeRepresentation>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,
@@ -48,15 +47,6 @@ pub fn resolve_metadata(
     mut models: IndexMap<Qualified<ModelName>, models::Model>,
     mut commands: IndexMap<Qualified<CommandName>, commands::Command>,
 ) -> Result<Metadata, Error> {
-    // resolve relationships
-    let object_types = resolve_relationships(
-        metadata_accessor,
-        data_connectors,
-        object_types,
-        &models,
-        &commands,
-    )?;
-
     // resolve command permissions
     // TODO: make this return values rather than blindly mutating it's inputs
     resolve_command_permissions(
@@ -95,68 +85,13 @@ pub fn resolve_metadata(
     })
 }
 
-/// resolve relationships
-/// returns updated `types` value
-fn resolve_relationships(
-    metadata_accessor: &open_dds::accessor::MetadataAccessor,
-    data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-    mut object_types: HashMap<
-        Qualified<CustomTypeName>,
-        type_permissions::ObjectTypeWithPermissions,
-    >,
-    models: &IndexMap<Qualified<ModelName>, models::Model>,
-    commands: &IndexMap<Qualified<CommandName>, commands::Command>,
-) -> Result<HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>, Error>
-{
-    for open_dds::accessor::QualifiedObject {
-        subgraph,
-        object: relationship,
-    } in &metadata_accessor.relationships
-    {
-        let qualified_relationship_source_type_name =
-            Qualified::new(subgraph.to_string(), relationship.source.to_owned());
-        let object_representation = object_types
-            .get_mut(&qualified_relationship_source_type_name)
-            .ok_or_else(|| Error::RelationshipDefinedOnUnknownType {
-                relationship_name: relationship.name.clone(),
-                type_name: qualified_relationship_source_type_name.clone(),
-            })?;
-
-        let resolved_relationship = resolve_relationship(
-            relationship,
-            subgraph,
-            models,
-            commands,
-            data_connectors,
-            object_representation,
-        )?;
-
-        if object_representation
-            .object_type
-            .relationships
-            .insert(
-                resolved_relationship.field_name.clone(),
-                resolved_relationship,
-            )
-            .is_some()
-        {
-            return Err(Error::DuplicateRelationshipInSourceType {
-                type_name: qualified_relationship_source_type_name,
-                relationship_name: relationship.name.clone(),
-            });
-        }
-    }
-
-    Ok(object_types)
-}
-
 /// resolve command permissions
 /// this currently works by mutating `commands`, let's change it to
 /// return new values instead where possible
 fn resolve_command_permissions(
     metadata_accessor: &open_dds::accessor::MetadataAccessor,
     commands: &mut IndexMap<Qualified<CommandName>, commands::Command>,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,
         boolean_expressions::ObjectBooleanExpressionType,
@@ -201,7 +136,7 @@ fn resolve_command_permissions(
 fn resolve_model_permissions(
     metadata_accessor: &open_dds::accessor::MetadataAccessor,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     models: &mut IndexMap<Qualified<ModelName>, models::Model>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,

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

@@ -1,8 +1,7 @@
 use super::permission::ValueExpression;
-use super::relationship::RelationshipTarget;
 use super::stages::{
     boolean_expressions, data_connector_scalar_types, data_connector_type_mappings, models,
-    type_permissions,
+    relationships,
 };
 use super::typecheck;
 
@@ -84,9 +83,8 @@ fn resolve_model_predicate(
     subgraph: &str,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
     fields: &IndexMap<FieldName, data_connector_type_mappings::FieldDefinition>,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     models: &IndexMap<Qualified<ModelName>, models::Model>,
-    // type_representation: &TypeRepresentation,
 ) -> Result<models::ModelPredicate, Error> {
     match model_predicate {
         permissions::ModelPredicate::FieldComparison(permissions::FieldComparisonPredicate {
@@ -213,7 +211,6 @@ fn resolve_model_predicate(
                 )?;
                 let relationship_field_name = mk_name(&name.0)?;
                 let relationship = &object_type_representation
-                    .object_type
                     .relationships
                     .get(&relationship_field_name)
                     .ok_or_else(|| Error::UnknownRelationshipInSelectPermissionsPredicate {
@@ -223,11 +220,13 @@ fn resolve_model_predicate(
                     })?;
 
                 match &relationship.target {
-                    RelationshipTarget::Command { .. } => Err(Error::UnsupportedFeature {
+                    relationships::RelationshipTarget::Command { .. } => {
+                        Err(Error::UnsupportedFeature {
                             message: "Predicate cannot be built using command relationships"
                                 .to_string(),
-                    }),
+                        })
-                    RelationshipTarget::Model {
+                    }
+                    relationships::RelationshipTarget::Model {
                         model_name,
                         relationship_type,
                         target_typename,
@@ -364,7 +363,7 @@ pub fn resolve_model_select_permissions(
     subgraph: &str,
     model_permissions: &ModelPermissionsV1,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     models: &IndexMap<Qualified<ModelName>, models::Model>,
     boolean_expression_types: &HashMap<
         Qualified<CustomTypeName>,
@@ -505,11 +504,11 @@ fn resolve_binary_operator_for_model(
 fn get_model_object_type_representation<'s>(
     object_types: &'s HashMap<
         Qualified<CustomTypeName>,
-        type_permissions::ObjectTypeWithPermissions,
+        relationships::ObjectTypeWithRelationships,
     >,
     data_type: &Qualified<CustomTypeName>,
     model_name: &Qualified<ModelName>,
-) -> Result<&'s type_permissions::ObjectTypeWithPermissions, crate::metadata::resolved::error::Error>
+) -> Result<&'s relationships::ObjectTypeWithRelationships, crate::metadata::resolved::error::Error>
 {
     match object_types.get(data_type) {
         Some(object_type_representation) => Ok(object_type_representation),

v3/crates/engine/src/metadata/resolved/stages/data_connector_type_mappings/mod.rs

@@ -228,7 +228,6 @@ pub fn resolve_object_type(
 
     Ok(ObjectTypeRepresentation {
         fields: resolved_fields,
-        relationships: IndexMap::new(),
         global_id_fields: resolved_global_id_fields,
         graphql_output_type_name: graphql_type_name,
         graphql_input_type_name,

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

@@ -1,7 +1,5 @@
 use crate::metadata::resolved::error::Error;
-use crate::metadata::resolved::relationship::Relationship;
 use crate::metadata::resolved::subgraph::QualifiedTypeReference;
-
 use indexmap::IndexMap;
 
 use open_dds::types::{CustomTypeName, Deprecated, FieldName};
@@ -92,7 +90,6 @@ pub struct DataConnectorTypeMappingsOutput {
 #[display(fmt = "Display")]
 pub struct ObjectTypeRepresentation {
     pub fields: IndexMap<FieldName, FieldDefinition>,
-    pub relationships: IndexMap<ast::Name, Relationship>,
     pub global_id_fields: Vec<FieldName>,
     pub apollo_federation_config: Option<ResolvedObjectApolloFederationConfig>,
     pub graphql_output_type_name: Option<ast::TypeName>,

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

@@ -7,6 +7,7 @@ pub mod data_connector_type_mappings;
 pub mod data_connectors;
 pub mod graphql_config;
 pub mod models;
+pub mod relationships;
 pub mod roles;
 pub mod scalar_types;
 pub mod type_permissions;
@@ -97,11 +98,19 @@ pub fn resolve(metadata: open_dds::Metadata) -> Result<Metadata, Error> {
         &apollo_federation_entity_enabled_types,
     )?;
 
+    let object_types_with_relationships = relationships::resolve(
+        &metadata_accessor,
+        &data_connectors,
+        &object_types_with_permissions,
+        &models,
+        &commands,
+    )?;
+
     resolve_metadata(
         &metadata_accessor,
         &graphql_config,
         &data_connector_type_mappings,
-        object_types_with_permissions,
+        object_types_with_relationships,
         &scalar_types,
         &boolean_expression_types,
         &data_connectors,

v3/crates/engine/src/metadata/resolved/stages/relationships/mod.rs

@@ -1,89 +1,100 @@
-use super::error::{Error, RelationshipError};
+mod types;
-use super::stages::{
+pub use types::{
-    commands, data_connector_scalar_types, data_connectors, models, type_permissions,
+    ObjectTypeWithRelationships, Relationship, RelationshipCapabilities,
+    RelationshipCommandMapping, RelationshipExecutionCategory, RelationshipModelMapping,
+    RelationshipTarget, RelationshipTargetName,
 };
-use super::subgraph::Qualified;
-use super::subgraph::QualifiedTypeReference;
-use super::types::mk_name;
-use super::types::NdcColumnForComparison;
-use indexmap::IndexMap;
-use lang_graphql::ast::common as ast;
-use open_dds::arguments::ArgumentName;
-use open_dds::commands::CommandName;
 
-use open_dds::models::ModelName;
+use std::collections::HashMap;
-use open_dds::relationships::{
+
-    self, FieldAccess, RelationshipName, RelationshipType, RelationshipV1,
+use indexmap::IndexMap;
+
+use open_dds::{commands::CommandName, models::ModelName, types::CustomTypeName};
+
+use crate::metadata::resolved::error::{Error, RelationshipError};
+use crate::metadata::resolved::subgraph::Qualified;
+
+use crate::metadata::resolved::stages::{
+    commands, data_connector_scalar_types, data_connector_type_mappings, data_connectors, models,
+    type_permissions,
 };
-use open_dds::types::CustomTypeName;
+use crate::metadata::resolved::types::mk_name;
-use open_dds::types::Deprecated;
+
-use serde::{Deserialize, Serialize};
+use open_dds::relationships::{self, FieldAccess, RelationshipName, RelationshipV1};
 
 use std::collections::HashSet;
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+/// resolve relationships
-pub enum RelationshipTarget {
+/// returns updated `types` value
-    Model {
+pub fn resolve(
-        // TODO(Abhinav): Refactor resolved types to contain denormalized data (eg: actual resolved model)
+    metadata_accessor: &open_dds::accessor::MetadataAccessor,
-        model_name: Qualified<ModelName>,
+    data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-        relationship_type: RelationshipType,
+    object_types_with_permissions: &HashMap<
-        target_typename: Qualified<CustomTypeName>,
+        Qualified<CustomTypeName>,
-        mappings: Vec<RelationshipModelMapping>,
+        type_permissions::ObjectTypeWithPermissions,
+    >,
+    models: &IndexMap<Qualified<ModelName>, models::Model>,
+    commands: &IndexMap<Qualified<CommandName>, commands::Command>,
+) -> Result<HashMap<Qualified<CustomTypeName>, ObjectTypeWithRelationships>, Error> {
+    let mut object_types_with_relationships = HashMap::new();
+    for (
+        object_type_name,
+        type_permissions::ObjectTypeWithPermissions {
+            type_output_permissions,
+            type_input_permissions,
+            object_type,
         },
-    Command {
+    ) in object_types_with_permissions
-        command_name: Qualified<CommandName>,
+    {
-        target_type: QualifiedTypeReference,
+        object_types_with_relationships.insert(
-        mappings: Vec<RelationshipCommandMapping>,
+            object_type_name.clone(),
+            ObjectTypeWithRelationships {
+                object_type: object_type.clone(),
+                type_output_permissions: type_output_permissions.clone(),
+                type_input_permissions: type_input_permissions.clone(),
+                relationships: IndexMap::new(),
             },
-}
+        );
+    }
+    for open_dds::accessor::QualifiedObject {
+        subgraph,
+        object: relationship,
+    } in &metadata_accessor.relationships
+    {
+        let qualified_relationship_source_type_name =
+            Qualified::new(subgraph.to_string(), relationship.source.to_owned());
+        let object_representation = object_types_with_relationships
+            .get_mut(&qualified_relationship_source_type_name)
+            .ok_or_else(|| Error::RelationshipDefinedOnUnknownType {
+                relationship_name: relationship.name.clone(),
+                type_name: qualified_relationship_source_type_name.clone(),
+            })?;
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+        let resolved_relationship = resolve_relationship(
-pub enum RelationshipTargetName {
+            relationship,
-    Model(Qualified<ModelName>),
+            subgraph,
-    Command(Qualified<CommandName>),
+            models,
-}
+            commands,
+            data_connectors,
+            &object_representation.object_type,
+        )?;
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+        if object_representation
-pub struct RelationshipModelMapping {
+            .relationships
-    pub source_field: FieldAccess,
+            .insert(
-    pub target_field: FieldAccess,
+                resolved_relationship.field_name.clone(),
-    // Optional because we allow building schema without specifying a data source
+                resolved_relationship,
-    pub target_ndc_column: Option<NdcColumnForComparison>,
+            )
-}
+            .is_some()
+        {
+            return Err(Error::DuplicateRelationshipInSourceType {
+                type_name: qualified_relationship_source_type_name,
+                relationship_name: relationship.name.clone(),
+            });
+        }
+    }
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+    Ok(object_types_with_relationships)
-pub struct RelationshipCommandMapping {
-    pub source_field: FieldAccess,
-    pub argument_name: ArgumentName,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct Relationship {
-    pub name: RelationshipName,
-    // `ast::Name` representation of `RelationshipName`. This is used to avoid
-    // the recurring conversion between `RelationshipName` to `ast::Name` during
-    // relationship IR generation
-    pub field_name: ast::Name,
-    pub source: Qualified<CustomTypeName>,
-    pub target: RelationshipTarget,
-    pub target_capabilities: Option<RelationshipCapabilities>,
-    pub description: Option<String>,
-    pub deprecated: Option<Deprecated>,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
-pub struct RelationshipCapabilities {
-    // TODO: We don't handle relationships without foreach.
-    // Change this to a bool, when we support that
-    pub foreach: (),
-    pub relationships: bool,
-}
-
-pub enum RelationshipExecutionCategory {
-    // Push down relationship definition to the data connector
-    Local,
-    // Use foreach in the data connector to fetch related rows for multiple objects in a single request
-    RemoteForEach,
 }
 
 #[allow(clippy::match_single_binding)]
@@ -110,7 +121,7 @@ pub fn relationship_execution_category(
 fn resolve_relationship_source_mapping<'a>(
     relationship_name: &'a RelationshipName,
     source_type_name: &'a Qualified<CustomTypeName>,
-    source_type: &type_permissions::ObjectTypeWithPermissions,
+    source_type: &data_connector_type_mappings::ObjectTypeRepresentation,
     relationship_mapping: &'a open_dds::relationships::RelationshipMapping,
 ) -> Result<&'a FieldAccess, Error> {
     match &relationship_mapping.source {
@@ -125,11 +136,7 @@ fn resolve_relationship_source_mapping<'a>(
                 relationship_name: relationship_name.clone(),
             }),
             [field_access] => {
-                if !source_type
+                if !source_type.fields.contains_key(&field_access.field_name) {
-                    .object_type
-                    .fields
-                    .contains_key(&field_access.field_name)
-                {
                     return Err(Error::RelationshipError {
                         relationship_error:
                             RelationshipError::UnknownSourceFieldInRelationshipMapping {
@@ -151,7 +158,7 @@ fn resolve_relationship_source_mapping<'a>(
 fn resolve_relationship_mappings_model(
     relationship: &RelationshipV1,
     source_type_name: &Qualified<CustomTypeName>,
-    source_type: &type_permissions::ObjectTypeWithPermissions,
+    source_type: &data_connector_type_mappings::ObjectTypeRepresentation,
     target_model: &models::Model,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
 ) -> Result<Vec<RelationshipModelMapping>, Error> {
@@ -255,7 +262,7 @@ fn resolve_relationship_mappings_model(
 fn resolve_relationship_mappings_command(
     relationship: &RelationshipV1,
     source_type_name: &Qualified<CustomTypeName>,
-    source_type: &type_permissions::ObjectTypeWithPermissions,
+    source_type: &data_connector_type_mappings::ObjectTypeRepresentation,
     target_command: &commands::Command,
 ) -> Result<Vec<RelationshipCommandMapping>, Error> {
     let mut resolved_relationship_mappings = Vec::new();
@@ -382,7 +389,7 @@ pub fn resolve_relationship(
     models: &IndexMap<Qualified<ModelName>, models::Model>,
     commands: &IndexMap<Qualified<CommandName>, commands::Command>,
     data_connectors: &data_connector_scalar_types::DataConnectorsWithScalars,
-    source_type: &type_permissions::ObjectTypeWithPermissions,
+    source_type: &data_connector_type_mappings::ObjectTypeRepresentation,
 ) -> Result<Relationship, Error> {
     let source_type_name = Qualified::new(subgraph.to_string(), relationship.source.clone());
     let (relationship_target, source_data_connector, target_name) = match &relationship.target {

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

@@ -0,0 +1,93 @@
+use crate::metadata::resolved::stages::{data_connector_type_mappings, type_permissions};
+use crate::metadata::resolved::subgraph::{Qualified, QualifiedTypeReference};
+use indexmap::IndexMap;
+use open_dds::permissions::Role;
+use open_dds::{commands::CommandName, models::ModelName, types::CustomTypeName};
+use serde::{Deserialize, Serialize};
+
+use crate::metadata::resolved::types::NdcColumnForComparison;
+use lang_graphql::ast::common as ast;
+use open_dds::arguments::ArgumentName;
+use std::collections::HashMap;
+
+use open_dds::relationships::{FieldAccess, RelationshipName, RelationshipType};
+use open_dds::types::Deprecated;
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, derive_more::Display)]
+#[display(fmt = "Display")]
+pub struct ObjectTypeWithRelationships {
+    pub object_type: data_connector_type_mappings::ObjectTypeRepresentation,
+    /// permissions on this type, when it is used in an output context (e.g. as
+    /// a return type of Model or Command)
+    pub type_output_permissions: HashMap<Role, open_dds::permissions::TypeOutputPermission>,
+    /// permissions on this type, when it is used in an input context (e.g. in
+    /// an argument type of Model or Command)
+    pub type_input_permissions: HashMap<Role, type_permissions::TypeInputPermission>,
+    /// any relationships defined on this object
+    pub relationships: IndexMap<ast::Name, Relationship>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub enum RelationshipTarget {
+    Model {
+        // TODO(Abhinav): Refactor resolved types to contain denormalized data (eg: actual resolved model)
+        model_name: Qualified<ModelName>,
+        relationship_type: RelationshipType,
+        target_typename: Qualified<CustomTypeName>,
+        mappings: Vec<RelationshipModelMapping>,
+    },
+    Command {
+        command_name: Qualified<CommandName>,
+        target_type: QualifiedTypeReference,
+        mappings: Vec<RelationshipCommandMapping>,
+    },
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub enum RelationshipTargetName {
+    Model(Qualified<ModelName>),
+    Command(Qualified<CommandName>),
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct RelationshipModelMapping {
+    pub source_field: FieldAccess,
+    pub target_field: FieldAccess,
+    // Optional because we allow building schema without specifying a data source
+    pub target_ndc_column: Option<NdcColumnForComparison>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct RelationshipCommandMapping {
+    pub source_field: FieldAccess,
+    pub argument_name: ArgumentName,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct Relationship {
+    pub name: RelationshipName,
+    // `ast::Name` representation of `RelationshipName`. This is used to avoid
+    // the recurring conversion between `RelationshipName` to `ast::Name` during
+    // relationship IR generation
+    pub field_name: ast::Name,
+    pub source: Qualified<CustomTypeName>,
+    pub target: RelationshipTarget,
+    pub target_capabilities: Option<RelationshipCapabilities>,
+    pub description: Option<String>,
+    pub deprecated: Option<Deprecated>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct RelationshipCapabilities {
+    // TODO: We don't handle relationships without foreach.
+    // Change this to a bool, when we support that
+    pub foreach: (),
+    pub relationships: bool,
+}
+
+pub enum RelationshipExecutionCategory {
+    // Push down relationship definition to the data connector
+    Local,
+    // Use foreach in the data connector to fetch related rows for multiple objects in a single request
+    RemoteForEach,
+}

v3/crates/engine/src/metadata/resolved/stages/roles/mod.rs

@@ -7,11 +7,11 @@ use open_dds::{commands::CommandName, models::ModelName, types::CustomTypeName};
 
 use crate::metadata::resolved::subgraph::Qualified;
 
-use crate::metadata::resolved::stages::{commands, models, type_permissions};
+use crate::metadata::resolved::stages::{commands, models, relationships};
 
 /// Gather all roles from various permission objects.
 pub fn resolve(
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
     models: &IndexMap<Qualified<ModelName>, models::Model>,
     commands: &IndexMap<Qualified<CommandName>, commands::Command>,
 ) -> Vec<Role> {

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

@@ -1,5 +1,6 @@
 use super::stages::{
-    boolean_expressions, data_connector_type_mappings, scalar_types, type_permissions,
+    boolean_expressions, data_connector_type_mappings, relationships, scalar_types,
+    type_permissions,
 };
 use crate::metadata::resolved::error::{BooleanExpressionError, Error};
 
@@ -53,26 +54,23 @@ pub fn resolve_field(
 #[derive(Debug)]
 /// we do not want to store our types like this, but occasionally it is useful
 /// for pattern matching
-pub enum TypeRepresentation<'a> {
+pub enum TypeRepresentation<'a, ObjectType> {
     Scalar(&'a scalar_types::ScalarTypeRepresentation),
-    Object(&'a type_permissions::ObjectTypeWithPermissions),
+    Object(&'a ObjectType),
     BooleanExpression(&'a boolean_expressions::ObjectBooleanExpressionType),
 }
 
 /// validate whether a given CustomTypeName exists within `object_types`, `scalar_types` or
 /// `boolean_expression_types`
-pub fn get_type_representation<'a>(
+pub fn get_type_representation<'a, ObjectType>(
     custom_type_name: &Qualified<CustomTypeName>,
-    object_types: &'a HashMap<
+    object_types: &'a HashMap<Qualified<CustomTypeName>, ObjectType>,
-        Qualified<CustomTypeName>,
-        type_permissions::ObjectTypeWithPermissions,
-    >,
     scalar_types: &'a HashMap<Qualified<CustomTypeName>, scalar_types::ScalarTypeRepresentation>,
     boolean_expression_types: &'a HashMap<
         Qualified<CustomTypeName>,
         boolean_expressions::ObjectBooleanExpressionType,
     >,
-) -> Result<TypeRepresentation<'a>, Error> {
+) -> Result<TypeRepresentation<'a, ObjectType>, Error> {
     match object_types.get(custom_type_name) {
         Some(object_type_representation) => {
             Ok(TypeRepresentation::Object(object_type_representation))
@@ -97,9 +95,9 @@ pub(crate) fn get_object_type_for_boolean_expression<'a>(
     boolean_expression_type: &boolean_expressions::ObjectBooleanExpressionType,
     object_types: &'a HashMap<
         Qualified<CustomTypeName>,
-        type_permissions::ObjectTypeWithPermissions,
+        relationships::ObjectTypeWithRelationships,
     >,
-) -> Result<&'a type_permissions::ObjectTypeWithPermissions, Error> {
+) -> Result<&'a relationships::ObjectTypeWithRelationships, Error> {
     object_types
         .get(&boolean_expression_type.object_type)
         .ok_or(Error::from(
@@ -112,9 +110,9 @@ pub(crate) fn get_object_type_for_boolean_expression<'a>(
 // Get the underlying object type by resolving Custom ObjectType, Array and
 // Nullable container types
 // check that `custom_type_name` exists in `object_types`
-pub fn object_type_exists(
+pub fn object_type_exists<ObjectType>(
     custom_type_name: &Qualified<CustomTypeName>,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, ObjectType>,
 ) -> Result<Qualified<CustomTypeName>, Error> {
     object_types
         .get(custom_type_name)

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

@@ -8,9 +8,6 @@ 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;
 
@@ -143,9 +140,8 @@ pub fn build_boolean_expression_input_schema(
 
         // relationship fields
         // TODO(naveen): Add support for command relationships
-        for (rel_name, relationship) in object_type_representation.object_type.relationships.iter()
+        for (rel_name, relationship) in object_type_representation.relationships.iter() {
-        {
+            if let resolved::RelationshipTarget::Model {
-            if let RelationshipTarget::Model {
                 model_name,
                 relationship_type,
                 target_typename,
@@ -168,11 +164,13 @@ pub fn build_boolean_expression_input_schema(
                         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(
+                    if let resolved::RelationshipExecutionCategory::Local =
+                        resolved::relationship_execution_category(
                             &boolean_expression_type.data_connector_link,
                             &target_source.data_connector,
                             &target_model_source.capabilities,
-                    ) {
+                        )
+                    {
                         if target_source.data_connector.name
                             == boolean_expression_type.data_connector_name
                         {

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

@@ -8,9 +8,6 @@ use std::collections::{BTreeMap, HashMap};
 use super::types::output_type::relationship::{ModelTargetSource, OrderByRelationshipAnnotation};
 use super::types::{output_type::get_object_type_representation, Annotation, 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;
@@ -159,9 +156,8 @@ pub fn build_model_order_by_input_schema(
 
         // relationship fields
         // TODO(naveen): Add support for command relationships.
-        for (rel_name, relationship) in object_type_representation.object_type.relationships.iter()
+        for (rel_name, relationship) in object_type_representation.relationships.iter() {
-        {
+            if let resolved::RelationshipTarget::Model {
-            if let RelationshipTarget::Model {
                 model_name,
                 relationship_type,
                 target_typename,
@@ -185,11 +181,13 @@ pub fn build_model_order_by_input_schema(
                     let target_model_source =
                         ModelTargetSource::from_model_source(target_source, relationship)?;
                     // order_by expression with relationships is currently only supported for local relationships
-                    if let RelationshipExecutionCategory::Local = relationship_execution_category(
+                    if let resolved::RelationshipExecutionCategory::Local =
+                        resolved::relationship_execution_category(
                             &model_source.data_connector,
                             &target_source.data_connector,
                             &target_model_source.capabilities,
-                    ) {
+                        )
+                    {
                         // TODO(naveen): Support Array relationships in order_by when the support for aggregates is implemented
                         if let RelationshipType::Object = relationship_type {
                             // If the relationship target model does not have orderByExpressionType do not include

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

@@ -2,7 +2,9 @@ use open_dds::types::{CustomTypeName, FieldName};
 use std::collections::{BTreeMap, HashMap};
 
 use crate::metadata::resolved::permission::ValueExpression;
-use crate::metadata::resolved::stages::{data_connector_type_mappings, models, type_permissions};
+use crate::metadata::resolved::stages::{
+    data_connector_type_mappings, models, relationships, type_permissions,
+};
 use crate::metadata::resolved::subgraph::{Qualified, QualifiedTypeReference};
 use crate::metadata::resolved::types::{object_type_exists, unwrap_custom_type_name};
 use crate::metadata::resolved::{self};
@@ -14,7 +16,7 @@ use super::types::ArgumentNameAndPath;
 /// Build namespace annotation for select permissions
 pub(crate) fn get_select_permissions_namespace_annotations(
     model: &models::Model,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
 ) -> Result<HashMap<Role, Option<types::NamespaceAnnotation>>, schema::Error> {
     let mut permissions: HashMap<Role, Option<types::NamespaceAnnotation>> = model
         .select_permissions
@@ -97,9 +99,9 @@ pub(crate) fn get_select_permissions_namespace_annotations(
 /// as we need to check the permissions of the arguments used in the selection.
 pub(crate) fn get_select_one_namespace_annotations(
     model: &models::Model,
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
     select_unique: &models::SelectUniqueGraphQlDefinition,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
 ) -> Result<HashMap<Role, Option<types::NamespaceAnnotation>>, schema::Error> {
     let select_permissions = get_select_permissions_namespace_annotations(model, object_types)?;
 
@@ -120,10 +122,10 @@ pub(crate) fn get_select_one_namespace_annotations(
 /// in the relationship mappings.
 pub(crate) fn get_model_relationship_namespace_annotations(
     target_model: &models::Model,
-    source_object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    source_object_type_representation: &resolved::ObjectTypeWithRelationships,
-    target_object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    target_object_type_representation: &resolved::ObjectTypeWithRelationships,
-    mappings: &[resolved::relationship::RelationshipModelMapping],
+    mappings: &[resolved::RelationshipModelMapping],
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, relationships::ObjectTypeWithRelationships>,
 ) -> Result<HashMap<Role, Option<types::NamespaceAnnotation>>, schema::Error> {
     let select_permissions =
         get_select_permissions_namespace_annotations(target_model, object_types)?;
@@ -147,7 +149,7 @@ pub(crate) fn get_model_relationship_namespace_annotations(
 /// Build namespace annotation for commands
 pub(crate) fn get_command_namespace_annotations(
     command: &resolved::Command,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
 ) -> Result<HashMap<Role, Option<types::NamespaceAnnotation>>, crate::schema::Error> {
     let mut permissions = HashMap::new();
 
@@ -224,16 +226,16 @@ fn build_annotations_from_input_object_type_permissions(
     field_path: &mut [String],
     type_reference: &QualifiedTypeReference,
     ndc_argument_name: &Option<String>,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
     type_mappings: &BTreeMap<Qualified<CustomTypeName>, data_connector_type_mappings::TypeMapping>,
     role_presets_map: &mut HashMap<Role, types::ArgumentPresets>,
 ) -> Result<(), schema::Error> {
-    if let Some(custom_typename) = unwrap_custom_type_name(type_reference) {
+    if let Some(object_type) = unwrap_custom_type_name(type_reference) {
-        if let Ok(object_type) = object_type_exists(custom_typename, object_types) {
+        if object_type_exists(object_type, object_types).is_ok() {
-            if let Some(object_type_repr) = object_types.get(&object_type) {
+            if let Some(object_type_repr) = object_types.get(object_type) {
                 let field_mappings =
                     type_mappings
-                        .get(&object_type)
+                        .get(object_type)
                         .map(|type_mapping| match type_mapping {
                             data_connector_type_mappings::TypeMapping::Object {
                                 ndc_object_type_name: _,
@@ -248,7 +250,7 @@ fn build_annotations_from_input_object_type_permissions(
                         type_reference,
                         field_path,
                         ndc_argument_name,
-                        &object_type,
+                        object_type,
                     )?;
 
                     role_presets_map.insert(
@@ -359,9 +361,9 @@ fn build_preset_map_from_input_object_type_permission(
 /// in the relationship mappings.
 pub(crate) fn get_command_relationship_namespace_annotations(
     command: &resolved::Command,
-    source_object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    source_object_type_representation: &resolved::ObjectTypeWithRelationships,
-    mappings: &[resolved::relationship::RelationshipCommandMapping],
+    mappings: &[resolved::RelationshipCommandMapping],
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
 ) -> Result<HashMap<Role, Option<types::NamespaceAnnotation>>, crate::schema::Error> {
     let select_permissions = get_command_namespace_annotations(command, object_types)?;
 
@@ -384,7 +386,7 @@ pub(crate) fn get_command_relationship_namespace_annotations(
 /// for a role if the role has access (select permissions)
 /// to all the Global ID fields.
 pub(crate) fn get_node_interface_annotations(
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
 ) -> HashMap<Role, Option<types::NamespaceAnnotation>> {
     let mut permissions = HashMap::new();
     for (role, type_output_permission) in &object_type_representation.type_output_permissions {
@@ -405,7 +407,7 @@ pub(crate) fn get_node_interface_annotations(
 /// for a role if the role has access (select permissions)
 /// to all the key fields.
 pub(crate) fn get_entity_union_permissions(
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
 ) -> HashMap<Role, Option<types::NamespaceAnnotation>> {
     let mut permissions = HashMap::new();
     for (role, type_output_permission) in &object_type_representation.type_output_permissions {
@@ -423,7 +425,7 @@ pub(crate) fn get_entity_union_permissions(
 
 /// Build namespace annotations for each field based on the type permissions
 pub(crate) fn get_allowed_roles_for_field<'a>(
-    object_type_representation: &'a type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &'a resolved::ObjectTypeWithRelationships,
     field_name: &'a FieldName,
 ) -> impl Iterator<Item = &'a Role> {
     object_type_representation
@@ -440,7 +442,7 @@ pub(crate) fn get_allowed_roles_for_field<'a>(
 
 /// Builds namespace annotations for the `node` field.
 pub(crate) fn get_node_field_namespace_permissions(
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
     model: &models::Model,
 ) -> HashMap<Role, resolved::FilterPermission> {
     let mut permissions = HashMap::new();
@@ -479,7 +481,7 @@ pub(crate) fn get_node_field_namespace_permissions(
 
 /// Builds namespace annotations for the `_entities` field.
 pub(crate) fn get_entities_field_namespace_permissions(
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
     model: &models::Model,
 ) -> HashMap<Role, resolved::FilterPermission> {
     let mut permissions = HashMap::new();

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

@@ -1,6 +1,7 @@
 use crate::{
     metadata::resolved::{
-        stages::{data_connector_type_mappings, type_permissions},
+        self,
+        stages::data_connector_type_mappings,
         subgraph::{Qualified, QualifiedBaseType, QualifiedTypeName, QualifiedTypeReference},
         types::{get_type_representation, mk_name, TypeRepresentation},
     },
@@ -80,7 +81,7 @@ fn get_custom_input_type(
     .map_err(|_| crate::schema::Error::InternalTypeNotFound {
         type_name: gds_type_name.clone(),
     })? {
-        TypeRepresentation::Object(type_permissions::ObjectTypeWithPermissions {
+        TypeRepresentation::Object(resolved::ObjectTypeWithRelationships {
             object_type:
                 data_connector_type_mappings::ObjectTypeRepresentation {
                     graphql_input_type_name,
@@ -125,7 +126,7 @@ fn get_custom_input_type(
 fn input_object_type_input_fields(
     gds: &GDS,
     builder: &mut gql_schema::Builder<GDS>,
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
 ) -> Result<BTreeMap<ast::Name, gql_schema::Namespaced<GDS, gql_schema::InputField<GDS>>>, Error> {
     object_type_representation
         .object_type

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

@@ -14,7 +14,7 @@ use self::relationship::{
 };
 use super::inbuilt_type::base_type_container_for_inbuilt_type;
 use super::{Annotation, PossibleApolloFederationTypes, TypeId};
-use crate::metadata::resolved::stages::{data_connector_type_mappings, type_permissions};
+use crate::metadata::resolved::stages::data_connector_type_mappings;
 use crate::metadata::resolved::subgraph::{
     Qualified, QualifiedBaseType, QualifiedTypeName, QualifiedTypeReference,
 };
@@ -190,8 +190,8 @@ fn object_type_fields(
     gds: &GDS,
     builder: &mut gql_schema::Builder<GDS>,
     type_name: &Qualified<CustomTypeName>,
-    object_type_representation: &type_permissions::ObjectTypeWithPermissions,
+    object_type_representation: &resolved::ObjectTypeWithRelationships,
-    object_types: &HashMap<Qualified<CustomTypeName>, type_permissions::ObjectTypeWithPermissions>,
+    object_types: &HashMap<Qualified<CustomTypeName>, resolved::ObjectTypeWithRelationships>,
 ) -> Result<BTreeMap<ast::Name, gql_schema::Namespaced<GDS, gql_schema::Field<GDS>>>, Error> {
     let mut graphql_fields = object_type_representation
         .object_type
@@ -225,13 +225,11 @@ fn object_type_fields(
             Ok((graphql_field_name, namespaced_field))
         })
         .collect::<Result<BTreeMap<_, _>, _>>()?;
-    for (relationship_field_name, relationship) in
+    for (relationship_field_name, relationship) in &object_type_representation.relationships {
-        &object_type_representation.object_type.relationships
-    {
         let deprecation_status = mk_deprecation_status(&relationship.deprecated);
 
         let relationship_field = match &relationship.target {
-            resolved::relationship::RelationshipTarget::Command {
+            resolved::RelationshipTarget::Command {
                 command_name,
                 target_type,
                 mappings,
@@ -292,7 +290,7 @@ fn object_type_fields(
                     )?,
                 )
             }
-            resolved::relationship::RelationshipTarget::Model {
+            resolved::RelationshipTarget::Model {
                 model_name,
                 relationship_type,
                 target_typename,
@@ -516,7 +514,7 @@ pub fn output_type_schema(
 pub(crate) fn get_object_type_representation<'s>(
     gds: &'s GDS,
     gds_type: &Qualified<CustomTypeName>,
-) -> Result<&'s type_permissions::ObjectTypeWithPermissions, crate::schema::Error> {
+) -> Result<&'s resolved::ObjectTypeWithRelationships, crate::schema::Error> {
     gds.metadata.object_types.get(gds_type).ok_or_else(|| {
         crate::schema::Error::InternalTypeNotFound {
             type_name: gds_type.clone(),

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

@@ -32,7 +32,7 @@ pub struct ModelRelationshipAnnotation {
     pub target_source: Option<ModelTargetSource>,
     pub target_type: Qualified<CustomTypeName>,
     pub relationship_type: RelationshipType,
-    pub mappings: Vec<resolved::relationship::RelationshipModelMapping>,
+    pub mappings: Vec<resolved::RelationshipModelMapping>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -49,7 +49,7 @@ pub struct FilterRelationshipAnnotation {
     pub target_source: ModelTargetSource,
     pub target_type: Qualified<CustomTypeName>,
     pub target_model_name: Qualified<ModelName>,
-    pub mappings: Vec<resolved::relationship::RelationshipModelMapping>,
+    pub mappings: Vec<resolved::RelationshipModelMapping>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -66,7 +66,7 @@ pub struct OrderByRelationshipAnnotation {
     pub target_source: ModelTargetSource,
     pub target_type: Qualified<CustomTypeName>,
     pub target_model_name: Qualified<ModelName>,
-    pub mappings: Vec<resolved::relationship::RelationshipModelMapping>,
+    pub mappings: Vec<resolved::RelationshipModelMapping>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -83,19 +83,19 @@ pub struct PredicateRelationshipAnnotation {
     pub target_source: ModelTargetSource,
     pub target_type: Qualified<CustomTypeName>,
     pub target_model_name: Qualified<ModelName>,
-    pub mappings: Vec<resolved::relationship::RelationshipModelMapping>,
+    pub mappings: Vec<resolved::RelationshipModelMapping>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 pub struct ModelTargetSource {
     pub(crate) model: models::ModelSource,
-    pub(crate) capabilities: resolved::relationship::RelationshipCapabilities,
+    pub(crate) capabilities: resolved::RelationshipCapabilities,
 }
 
 impl ModelTargetSource {
     pub fn new(
         model: &models::Model,
-        relationship: &resolved::relationship::Relationship,
+        relationship: &resolved::Relationship,
     ) -> Result<Option<Self>, schema::Error> {
         model
             .source
@@ -106,7 +106,7 @@ impl ModelTargetSource {
 
     pub fn from_model_source(
         model_source: &models::ModelSource,
-        relationship: &resolved::relationship::Relationship,
+        relationship: &resolved::Relationship,
     ) -> Result<Self, schema::Error> {
         Ok(Self {
             model: model_source.clone(),
@@ -130,20 +130,20 @@ pub struct CommandRelationshipAnnotation {
     pub target_source: Option<CommandTargetSource>,
     pub target_type: QualifiedTypeReference,
     pub target_base_type_kind: TypeKind,
-    pub mappings: Vec<resolved::relationship::RelationshipCommandMapping>,
+    pub mappings: Vec<resolved::RelationshipCommandMapping>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 pub struct CommandTargetSource {
     pub(crate) details: CommandSourceDetail,
     pub(crate) function_name: FunctionName,
-    pub(crate) capabilities: resolved::relationship::RelationshipCapabilities,
+    pub(crate) capabilities: resolved::RelationshipCapabilities,
 }
 
 impl CommandTargetSource {
     pub fn new(
         command: &resolved::Command,
-        relationship: &resolved::relationship::Relationship,
+        relationship: &resolved::Relationship,
     ) -> Result<Option<Self>, schema::Error> {
         command
             .source