generate IR for type to command local relationships (#251)

Co-authored-by: Abhinav Gupta <abhinav@hasura.io>
V3_GIT_ORIGIN_REV_ID: b49b4c236c0df997ca8fcbe0a3e52e90f731f544

v3/custom-connector/src/bin/agent/main.rs

@@ -13,7 +13,7 @@ use axum::{
 };
 
 use indexmap::IndexMap;
-use ndc_client::models;
+use ndc_client::models::{self, Query};
 use prometheus::{Encoder, IntCounter, IntGauge, Opts, Registry, TextEncoder};
 use regex::Regex;
 use tokio::sync::Mutex;
@@ -533,6 +533,25 @@ async fn get_schema() -> Json<models::SchemaResponse> {
     };
     // ANCHOR_END: schema_function_get_actor_by_id
 
+    // ANCHOR: schema_function_get_movie_by_id
+    let get_movie_by_id_function = models::FunctionInfo {
+        name: "get_movie_by_id".into(),
+        description: Some("Get movie by ID".into()),
+        arguments: BTreeMap::from_iter([(
+            "movie_id".into(),
+            models::ArgumentInfo {
+                description: Some("the id of the movie to fetch".into()),
+                argument_type: models::Type::Named { name: "Int".into() },
+            },
+        )]),
+        result_type: models::Type::Nullable {
+            underlying_type: Box::new(models::Type::Named {
+                name: "movie".into(),
+            }),
+        },
+    };
+    // ANCHOR_END: schema_function_get_movie_by_id
+
     // ANCHOR: schema_function_get_actors_by_name
     let get_actors_by_name_function = models::FunctionInfo {
         name: "get_actors_by_name".into(),
@@ -560,6 +579,7 @@ async fn get_schema() -> Json<models::SchemaResponse> {
         latest_actor_name_function,
         latest_actor_function,
         get_actor_by_id_function,
+        get_movie_by_id_function,
         get_actors_by_name_function,
     ];
     // ANCHOR_END: schema_functions
@@ -670,6 +690,9 @@ fn get_collection_by_name(
         "get_actor_by_id" => {
             get_actor_by_id_rows(arguments, state, query, collection_relationships, variables)
         }
+        "get_movie_by_id" => {
+            get_movie_by_id_rows(arguments, state, query, collection_relationships, variables)
+        }
         "get_actors_by_name" => {
             get_actors_by_name_rows(arguments, state, query, collection_relationships, variables)
         }
@@ -898,6 +921,55 @@ fn get_actor_by_id_rows(
     }
 }
 
+fn get_movie_by_id_rows(
+    arguments: &BTreeMap<String, serde_json::Value>,
+    state: &AppState,
+    query: &models::Query,
+    collection_relationships: &BTreeMap<String, models::Relationship>,
+    variables: &BTreeMap<String, serde_json::Value>,
+) -> Result<Vec<Row>> {
+    let id_value = arguments.get("movie_id").ok_or((
+        StatusCode::BAD_REQUEST,
+        Json(models::ErrorResponse {
+            message: "missing argument id".into(),
+            details: serde_json::Value::Null,
+        }),
+    ))?;
+    if let Some(id) = id_value.as_i64() {
+        let movie = state.movies.get(&id);
+
+        match movie {
+            None => Ok(vec![BTreeMap::from_iter([(
+                "__value".into(),
+                serde_json::Value::Null,
+            )])]),
+            Some(movie) => {
+                let rows = project_row(movie, state, query, collection_relationships, variables)?;
+
+                let movie_value = serde_json::to_value(rows).map_err(|_| {
+                    (
+                        StatusCode::INTERNAL_SERVER_ERROR,
+                        Json(models::ErrorResponse {
+                            message: "unable to encode value".into(),
+                            details: serde_json::Value::Null,
+                        }),
+                    )
+                })?;
+
+                Ok(vec![BTreeMap::from_iter([("__value".into(), movie_value)])])
+            }
+        }
+    } else {
+        Err((
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(models::ErrorResponse {
+                message: "incorrect type for id".into(),
+                details: serde_json::Value::Null,
+            }),
+        ))
+    }
+}
+
 fn project_row(
     row: &Row,
     state: &AppState,
@@ -905,7 +977,7 @@ fn project_row(
     collection_relationships: &BTreeMap<String, models::Relationship>,
     variables: &BTreeMap<String, serde_json::Value>,
 ) -> Result<Option<IndexMap<String, models::RowFieldValue>>> {
-    let row = query
+    query
         .fields
         .as_ref()
         .map(|fields| {
@@ -918,8 +990,7 @@ fn project_row(
                 })
                 .collect::<Result<IndexMap<String, models::RowFieldValue>>>()
         })
-        .transpose()?;
+        .transpose()
-    Ok(row)
 }
 
 fn get_actors_by_name_rows(
@@ -1337,6 +1408,7 @@ fn execute_query(
     // ANCHOR_END: execute_query_filter
     // ANCHOR: execute_query_paginate
     let paginated: Vec<Row> = paginate(filtered.into_iter(), query.limit, query.offset);
+
     // ANCHOR_END: execute_query_paginate
     // ANCHOR: execute_query_aggregates
     let aggregates = query
@@ -1395,6 +1467,7 @@ fn execute_query(
         .transpose()?;
     // ANCHOR_END: execute_query_fields
     // ANCHOR: execute_query_rowset
+
     Ok(models::RowSet { aggregates, rows })
     // ANCHOR_END: execute_query_rowset
 }
@@ -1729,6 +1802,7 @@ fn eval_path(
             relationship,
             &path_element.arguments,
             &result,
+            None,
             &path_element.predicate,
         )?;
     }
@@ -1744,6 +1818,7 @@ fn eval_path_element(
     relationship: &models::Relationship,
     arguments: &BTreeMap<String, models::RelationshipArgument>,
     source: &[Row],
+    query: Option<&Query>,
     predicate: &models::Expression,
 ) -> Result<Vec<Row>> {
     let mut matching_rows: Vec<Row> = vec![];
@@ -1804,13 +1879,16 @@ fn eval_path_element(
             }
         }
 
-        let query = models::Query {
+        let query = match query {
+            None => models::Query {
                 aggregates: None,
                 fields: Some(IndexMap::new()),
                 limit: None,
                 offset: None,
                 order_by: None,
                 predicate: None,
+            },
+            Some(query) => query.clone(),
         };
 
         let target = get_collection_by_name(
@@ -2145,6 +2223,7 @@ fn eval_in_collection(
                 relationship,
                 arguments,
                 &source,
+                None,
                 &models::Expression::And {
                     expressions: vec![],
                 },
@@ -2284,6 +2363,7 @@ fn eval_field(
                 relationship,
                 arguments,
                 &source,
+                Some(query),
                 &models::Expression::And {
                     expressions: vec![],
                 },

v3/engine/src/execute/error.rs

@@ -46,7 +46,6 @@ pub enum InternalDeveloperError {
         type_name: Qualified<CustomTypeName>,
         relationship_name: RelationshipName,
     },
-
     #[error("Field mapping not found for the field {field_name:} of type {type_name:} while executing the relationship {relationship_name:}")]
     FieldMappingNotFoundForRelationship {
         type_name: Qualified<CustomTypeName>,
@@ -87,6 +86,15 @@ pub enum InternalEngineError {
     #[error("remote relationships should have been handled separately")]
     RemoteRelationshipsAreNotSupported,
 
+    #[error("relationships to procedure based commands are not supported")]
+    RelationshipsToProcedureBasedCommandsAreNotSupported,
+
+    #[error("remote relationships to command are not supported")]
+    RemoteCommandRelationshipsAreNotSupported,
+
+    #[error("expected filter predicate but filter predicate namespaced annotation not found")]
+    FilterPermissionAnnotationNotFound,
+
     #[error("expected namespace annotation type {namespace_annotation_type} but not found")]
     // Running into this error means that the GDS field was not annotated with the correct
     // namespace annotation while building the metadata.

v3/engine/src/execute/ir/commands.rs

@@ -24,7 +24,7 @@ use crate::schema::GDS;
 
 /// IR for the 'command' operations
 #[derive(Serialize, Debug)]
-pub struct CommandRepresentation<'n, 's> {
+pub struct CommandRepresentation<'s> {
     /// The name of the command
     pub command_name: subgraph::Qualified<commands::CommandName>,
 
@@ -32,10 +32,10 @@ pub struct CommandRepresentation<'n, 's> {
     pub field_name: ast::Name,
 
     /// The data connector backing this model.
-    pub data_connector: resolved::data_connector::DataConnector,
+    pub data_connector: &'s resolved::data_connector::DataConnector,
 
     /// Source function/procedure in the data connector for this model
-    pub ndc_source: DataConnectorCommand,
+    pub ndc_source: &'s DataConnectorCommand,
 
     /// Arguments for the NDC table
     pub(crate) arguments: BTreeMap<String, json::Value>,
@@ -45,7 +45,7 @@ pub struct CommandRepresentation<'n, 's> {
 
     /// The Graphql base type for the output_type of command. Helps in deciding how
     /// the response from the NDC needs to be processed.
-    pub type_container: &'n TypeContainer<TypeName>,
+    pub type_container: TypeContainer<TypeName>,
 
     // All the models/commands used in the 'command' operation.
     pub(crate) usage_counts: UsagesCounts,
@@ -56,11 +56,11 @@ pub struct CommandRepresentation<'n, 's> {
 pub(crate) fn command_generate_ir<'n, 's>(
     command_name: &subgraph::Qualified<commands::CommandName>,
     field: &'n normalized_ast::Field<'s, GDS>,
-    field_call: &'s normalized_ast::FieldCall<'s, GDS>,
+    field_call: &'n normalized_ast::FieldCall<'s, GDS>,
     underlying_object_typename: &Option<subgraph::Qualified<open_dds::types::CustomTypeName>>,
     command_source: &'s resolved::command::CommandSource,
     session_variables: &SessionVariables,
-) -> Result<CommandRepresentation<'n, 's>, error::Error> {
+) -> Result<CommandRepresentation<'s>, error::Error> {
     let empty_field_mappings = BTreeMap::new();
     // No field mappings should exists if the resolved output type of command is
     // not a custom object type
@@ -112,21 +112,20 @@ pub(crate) fn command_generate_ir<'n, 's>(
     Ok(CommandRepresentation {
         command_name: command_name.clone(),
         field_name: field_call.name.clone(),
-        data_connector: command_source.data_connector.clone(),
+        data_connector: &command_source.data_connector,
-        ndc_source: command_source.source.clone(),
+        ndc_source: &command_source.source,
         arguments: command_arguments,
         selection,
-        type_container: &field.type_container,
+        type_container: field.type_container.clone(),
         // selection_set: &field.selection_set,
         usage_counts,
     })
 }
 
-pub fn ir_to_ndc_query_ir<'s>(
+pub fn ir_to_ndc_query<'s>(
-    function_name: &String,
+    ir: &CommandRepresentation<'s>,
-    ir: &CommandRepresentation<'_, 's>,
     join_id_counter: &mut MonotonicCounter,
-) -> Result<(gdc::models::QueryRequest, JoinLocations<RemoteJoin<'s>>), error::Error> {
+) -> Result<(gdc::models::Query, JoinLocations<RemoteJoin<'s>>), error::Error> {
     let (ndc_fields, jl) = selection_set::process_selection_set_ir(&ir.selection, join_id_counter)?;
     let query = gdc::models::Query {
         aggregates: None,
@@ -136,6 +135,15 @@ pub fn ir_to_ndc_query_ir<'s>(
         order_by: None,
         predicate: None,
     };
+    Ok((query, jl))
+}
+
+pub fn ir_to_ndc_query_ir<'s>(
+    function_name: &String,
+    ir: &CommandRepresentation<'s>,
+    join_id_counter: &mut MonotonicCounter,
+) -> Result<(gdc::models::QueryRequest, JoinLocations<RemoteJoin<'s>>), error::Error> {
+    let (query, jl) = ir_to_ndc_query(ir, join_id_counter)?;
     let mut collection_relationships = BTreeMap::new();
     selection_set::collect_relationships(&ir.selection, &mut collection_relationships)?;
     let arguments: BTreeMap<String, gdc::models::Argument> = ir
@@ -160,7 +168,7 @@ pub fn ir_to_ndc_query_ir<'s>(
 
 pub fn ir_to_ndc_mutation_ir<'s>(
     procedure_name: &String,
-    ir: &CommandRepresentation<'_, 's>,
+    ir: &CommandRepresentation<'s>,
     join_id_counter: &mut MonotonicCounter,
 ) -> Result<(gdc::models::MutationRequest, JoinLocations<RemoteJoin<'s>>), error::Error> {
     let arguments = ir

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

@@ -10,17 +10,22 @@ use open_dds::{
 use ndc_client as ndc;
 use serde::Serialize;
 
-use super::filter::resolve_filter_expression;
 use super::model_selection::model_selection_ir;
 use super::order_by::build_ndc_order_by;
 use super::permissions;
 use super::selection_set::FieldSelection;
-use crate::execute::error;
+use super::{commands::command_generate_ir, filter::resolve_filter_expression};
 use crate::execute::model_tracking::{count_model, UsagesCounts};
 use crate::metadata::resolved::subgraph::serialize_qualified_btreemap;
 use crate::schema::types::output_type::relationship::{
     ModelRelationshipAnnotation, ModelTargetSource,
 };
+use crate::{
+    execute::{error, model_tracking::count_command},
+    schema::types::output_type::relationship::{
+        CommandRelationshipAnnotation, CommandTargetSource,
+    },
+};
 use crate::{
     metadata::resolved::{self, subgraph::Qualified},
     schema::{
@@ -30,7 +35,7 @@ use crate::{
 };
 
 #[derive(Debug, Serialize)]
-pub(crate) struct RelationshipInfo<'s> {
+pub(crate) struct LocalModelRelationshipInfo<'s> {
     pub annotation: &'s ModelRelationshipAnnotation,
     pub source_data_connector: &'s resolved::data_connector::DataConnector,
     #[serde(serialize_with = "serialize_qualified_btreemap")]
@@ -38,8 +43,17 @@ pub(crate) struct RelationshipInfo<'s> {
     pub target_source: &'s ModelTargetSource,
 }
 
+#[derive(Debug, Serialize)]
+pub(crate) struct LocalCommandRelationshipInfo<'s> {
+    pub annotation: &'s CommandRelationshipAnnotation,
+    pub source_data_connector: &'s resolved::data_connector::DataConnector,
+    #[serde(serialize_with = "serialize_qualified_btreemap")]
+    pub source_type_mappings: &'s BTreeMap<Qualified<CustomTypeName>, resolved::types::TypeMapping>,
+    pub target_source: &'s CommandTargetSource,
+}
+
 #[derive(Debug, Clone, Serialize)]
-pub struct RemoteRelationshipInfo<'s> {
+pub struct RemoteModelRelationshipInfo<'s> {
     pub annotation: &'s ModelRelationshipAnnotation,
     /// This contains processed information about the mappings.
     /// `RelationshipMapping` only contains mapping of field names. This
@@ -51,10 +65,10 @@ pub struct RemoteRelationshipInfo<'s> {
 pub type SourceField = (FieldName, resolved::types::FieldMapping);
 pub type TargetField = (FieldName, resolved::types::FieldMapping);
 
-pub(crate) fn process_relationship_definition(
+pub(crate) fn process_model_relationship_definition(
-    relationship_info: &RelationshipInfo,
+    relationship_info: &LocalModelRelationshipInfo,
 ) -> Result<ndc::models::Relationship, error::Error> {
-    let &RelationshipInfo {
+    let &LocalModelRelationshipInfo {
         annotation,
         source_data_connector,
         source_type_mappings,
@@ -68,7 +82,11 @@ pub(crate) fn process_relationship_definition(
     } in annotation.mappings.iter()
     {
         if !matches!(
-            relationship_execution_category(target_source, source_data_connector),
+            relationship_execution_category(
+                source_data_connector,
+                &target_source.model.data_connector,
+                &target_source.capabilities
+            ),
             RelationshipExecutionCategory::Local
         ) {
             Err(error::InternalEngineError::RemoteRelationshipsAreNotSupported)?
@@ -111,6 +129,71 @@ pub(crate) fn process_relationship_definition(
     Ok(ndc_relationship)
 }
 
+pub(crate) fn process_command_relationship_definition(
+    relationship_info: &LocalCommandRelationshipInfo,
+) -> Result<ndc::models::Relationship, error::Error> {
+    let &LocalCommandRelationshipInfo {
+        annotation,
+        source_data_connector,
+        source_type_mappings,
+        target_source,
+    } = relationship_info;
+
+    let mut arguments = BTreeMap::new();
+    for resolved::relationship::RelationshipCommandMapping {
+        source_field: source_field_path,
+        argument_name: target_argument,
+    } in annotation.mappings.iter()
+    {
+        if !matches!(
+            relationship_execution_category(
+                source_data_connector,
+                &target_source.command.data_connector,
+                &target_source.capabilities
+            ),
+            RelationshipExecutionCategory::Local
+        ) {
+            Err(error::InternalEngineError::RemoteRelationshipsAreNotSupported)?
+        } else {
+            let source_column = get_field_mapping_of_field_name(
+                source_type_mappings,
+                &annotation.source_type,
+                &annotation.relationship_name,
+                &source_field_path.field_name,
+            )?;
+
+            let relationship_argument = ndc::models::RelationshipArgument::Column {
+                name: source_column.column,
+            };
+
+            if arguments
+                .insert(target_argument.to_string(), relationship_argument)
+                .is_some()
+            {
+                Err(error::InternalEngineError::MappingExistsInRelationship {
+                    source_column: source_field_path.field_name.clone(),
+                    relationship_name: annotation.relationship_name.clone(),
+                })?
+            }
+        }
+    }
+
+    let target_collection = match &target_source.command.source {
+        open_dds::commands::DataConnectorCommand::Function(function_name) => function_name,
+        open_dds::commands::DataConnectorCommand::Procedure(..) => {
+            Err(error::InternalEngineError::RelationshipsToProcedureBasedCommandsAreNotSupported)?
+        }
+    };
+
+    let ndc_relationship = ndc_client::models::Relationship {
+        column_mapping: BTreeMap::new(),
+        relationship_type: ndc_client::models::RelationshipType::Object,
+        target_collection: target_collection.to_string(),
+        arguments,
+    };
+    Ok(ndc_relationship)
+}
+
 enum RelationshipExecutionCategory {
     // Push down relationship definition to the data connector
     Local,
@@ -120,17 +203,16 @@ enum RelationshipExecutionCategory {
 
 #[allow(clippy::match_single_binding)]
 fn relationship_execution_category(
-    target_source: &ModelTargetSource,
     source_connector: &resolved::data_connector::DataConnector,
+    target_connector: &resolved::data_connector::DataConnector,
+    relationship_capabilities: &resolved::relationship::RelationshipCapabilities,
 ) -> RelationshipExecutionCategory {
     // It's a local relationship if the source and target connectors are the same and
     // the connector supports relationships.
-    if target_source.model.data_connector.name == source_connector.name
+    if target_connector.name == source_connector.name && relationship_capabilities.relationships {
-        && target_source.capabilities.relationships
-    {
         RelationshipExecutionCategory::Local
     } else {
-        match target_source.capabilities.foreach {
+        match relationship_capabilities.foreach {
             // TODO: When we support naive relationships for connectors not implementing foreach,
             // add another match arm / return enum variant
             () => RelationshipExecutionCategory::RemoteForEach,
@@ -138,10 +220,10 @@ fn relationship_execution_category(
     }
 }
 
-pub(crate) fn generate_relationship_ir<'s>(
+pub(crate) fn generate_model_relationship_ir<'s>(
     field: &Field<'s, GDS>,
     annotation: &'s ModelRelationshipAnnotation,
-    data_connector: &'s resolved::data_connector::DataConnector,
+    source_data_connector: &'s resolved::data_connector::DataConnector,
     type_mappings: &'s BTreeMap<Qualified<CustomTypeName>, resolved::types::TypeMapping>,
     session_variables: &SessionVariables,
     usage_counts: &mut UsagesCounts,
@@ -207,12 +289,16 @@ pub(crate) fn generate_relationship_ir<'s>(
                 }
                 None => error::Error::from(normalized_ast::Error::NoTypenameFound),
             })?;
-    match relationship_execution_category(target_source, data_connector) {
+    match relationship_execution_category(
-        RelationshipExecutionCategory::Local => build_local_relationship(
+        source_data_connector,
+        &target_source.model.data_connector,
+        &target_source.capabilities,
+    ) {
+        RelationshipExecutionCategory::Local => build_local_model_relationship(
             field,
             field_call,
             annotation,
-            data_connector,
+            source_data_connector,
             type_mappings,
             target_source,
             filter_clause,
@@ -238,8 +324,53 @@ pub(crate) fn generate_relationship_ir<'s>(
     }
 }
 
+pub(crate) fn generate_command_relationship_ir<'s>(
+    field: &Field<'s, GDS>,
+    annotation: &'s CommandRelationshipAnnotation,
+    source_data_connector: &'s resolved::data_connector::DataConnector,
+    type_mappings: &'s BTreeMap<Qualified<CustomTypeName>, resolved::types::TypeMapping>,
+    session_variables: &SessionVariables,
+    usage_counts: &mut UsagesCounts,
+) -> Result<FieldSelection<'s>, error::Error> {
+    count_command(annotation.command_name.clone(), usage_counts);
+    let field_call = field.field_call()?;
+
+    let target_source =
+        annotation
+            .target_source
+            .as_ref()
+            .ok_or_else(|| match &field.selection_set.type_name {
+                Some(type_name) => {
+                    error::Error::from(error::InternalDeveloperError::NoSourceDataConnector {
+                        type_name: type_name.clone(),
+                        field_name: field_call.name.clone(),
+                    })
+                }
+                None => error::Error::from(normalized_ast::Error::NoTypenameFound),
+            })?;
+
+    match relationship_execution_category(
+        source_data_connector,
+        &target_source.command.data_connector,
+        &target_source.capabilities,
+    ) {
+        RelationshipExecutionCategory::Local => build_local_command_relationship(
+            field,
+            field_call,
+            annotation,
+            source_data_connector,
+            type_mappings,
+            target_source,
+            session_variables,
+        ),
+        RelationshipExecutionCategory::RemoteForEach => {
+            Err(error::InternalEngineError::RemoteCommandRelationshipsAreNotSupported)?
+        }
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
-pub(crate) fn build_local_relationship<'s>(
+pub(crate) fn build_local_model_relationship<'s>(
     field: &normalized_ast::Field<'s, GDS>,
     field_call: &normalized_ast::FieldCall<'s, GDS>,
     annotation: &'s ModelRelationshipAnnotation,
@@ -266,7 +397,7 @@ pub(crate) fn build_local_relationship<'s>(
         session_variables,
         usage_counts,
     )?;
-    let rel_info = RelationshipInfo {
+    let rel_info = LocalModelRelationshipInfo {
         annotation,
         source_data_connector: data_connector,
         source_type_mappings: type_mappings,
@@ -282,13 +413,55 @@ pub(crate) fn build_local_relationship<'s>(
     let relationship_name =
         serde_json::to_string(&(&annotation.source_type, &annotation.relationship_name))?;
 
-    Ok(FieldSelection::LocalRelationship {
+    Ok(FieldSelection::ModelRelationshipLocal {
         query: relationships_ir,
         name: relationship_name,
         relationship_info: rel_info,
     })
 }
 
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn build_local_command_relationship<'s>(
+    field: &normalized_ast::Field<'s, GDS>,
+    field_call: &normalized_ast::FieldCall<'s, GDS>,
+    annotation: &'s CommandRelationshipAnnotation,
+    data_connector: &'s resolved::data_connector::DataConnector,
+    type_mappings: &'s BTreeMap<Qualified<CustomTypeName>, resolved::types::TypeMapping>,
+    target_source: &'s CommandTargetSource,
+    session_variables: &SessionVariables,
+) -> Result<FieldSelection<'s>, error::Error> {
+    let relationships_ir = command_generate_ir(
+        &annotation.command_name,
+        field,
+        field_call,
+        &annotation.underlying_object_typename,
+        &target_source.command,
+        session_variables,
+    )?;
+
+    let rel_info = LocalCommandRelationshipInfo {
+        annotation,
+        source_data_connector: data_connector,
+        source_type_mappings: type_mappings,
+        target_source,
+    };
+
+    // Relationship names needs to be unique across the IR. This is so that, the
+    // NDC can use these names to figure out what joins to use.
+    // A single "source type" can have only one relationship with a given name,
+    // hence the relationship name in the IR is a tuple between the source type
+    // and the relationship name.
+    // Relationship name = (source_type, relationship_name)
+    let relationship_name =
+        serde_json::to_string(&(&annotation.source_type, &annotation.relationship_name))?;
+
+    Ok(FieldSelection::CommandRelationshipLocal {
+        ir: relationships_ir,
+        name: relationship_name,
+        relationship_info: rel_info,
+    })
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn build_remote_relationship<'n, 's>(
     field: &'n normalized_ast::Field<'s, GDS>,
@@ -355,11 +528,11 @@ pub(crate) fn build_remote_relationship<'n, 's>(
         };
         remote_relationships_ir.filter_clause.push(comparison_exp);
     }
-    let rel_info = RemoteRelationshipInfo {
+    let rel_info = RemoteModelRelationshipInfo {
         annotation,
         join_mapping,
     };
-    Ok(FieldSelection::RemoteRelationship {
+    Ok(FieldSelection::ModelRelationshipRemote {
         ir: remote_relationships_ir,
         relationship_info: rel_info,
     })

v3/engine/src/execute/ir/root_field.rs

@@ -52,7 +52,7 @@ pub enum QueryRootField<'n, 's> {
     NodeSelect(Option<node_field::NodeSelect<'n, 's>>),
     CommandRepresentation {
         selection_set: &'n gql::normalized_ast::SelectionSet<'s, GDS>,
-        ir: commands::CommandRepresentation<'n, 's>,
+        ir: commands::CommandRepresentation<'s>,
     },
 }
 
@@ -65,6 +65,6 @@ pub enum MutationRootField<'n, 's> {
     },
     CommandRepresentation {
         selection_set: &'n gql::normalized_ast::SelectionSet<'s, GDS>,
-        ir: commands::CommandRepresentation<'n, 's>,
+        ir: commands::CommandRepresentation<'s>,
     },
 }

v3/engine/src/execute/ir/selection_set.rs

@@ -7,8 +7,11 @@ use open_dds::types::{CustomTypeName, FieldName};
 use serde::Serialize;
 use std::collections::{BTreeMap, HashMap};
 
+use super::commands::{self, CommandRepresentation};
 use super::model_selection::{self, ModelSelection};
-use super::relationship::{self, RelationshipInfo, RemoteRelationshipInfo};
+use super::relationship::{
+    self, LocalCommandRelationshipInfo, LocalModelRelationshipInfo, RemoteModelRelationshipInfo,
+};
 use crate::execute::error;
 use crate::execute::global_id;
 use crate::execute::model_tracking::UsagesCounts;
@@ -26,17 +29,22 @@ pub(crate) enum FieldSelection<'s> {
     Column {
         column: String,
     },
-    LocalRelationship {
+    ModelRelationshipLocal {
         query: ModelSelection<'s>,
         /// Relationship names needs to be unique across the IR. This field contains
         /// the uniquely generated relationship name. `ModelRelationshipAnnotation`
         /// contains a relationship name but that is the name from the metadata.
         name: String,
-        relationship_info: RelationshipInfo<'s>,
+        relationship_info: LocalModelRelationshipInfo<'s>,
     },
-    RemoteRelationship {
+    CommandRelationshipLocal {
+        ir: CommandRepresentation<'s>,
+        name: String,
+        relationship_info: LocalCommandRelationshipInfo<'s>,
+    },
+    ModelRelationshipRemote {
         ir: ModelSelection<'s>,
-        relationship_info: RemoteRelationshipInfo<'s>,
+        relationship_info: RemoteModelRelationshipInfo<'s>,
     },
 }
 
@@ -147,7 +155,20 @@ pub(crate) fn generate_selection_set_ir<'s>(
                 OutputAnnotation::RelationshipToModel(relationship_annotation) => {
                     fields.insert(
                         field.alias.to_string(),
-                        relationship::generate_relationship_ir(
+                        relationship::generate_model_relationship_ir(
+                            field,
+                            relationship_annotation,
+                            data_connector,
+                            type_mappings,
+                            session_variables,
+                            usage_counts,
+                        )?,
+                    );
+                }
+                OutputAnnotation::RelationshipToCommand(relationship_annotation) => {
+                    fields.insert(
+                        field.alias.to_string(),
+                        relationship::generate_command_relationship_ir(
                             field,
                             relationship_annotation,
                             data_connector,
@@ -193,7 +214,7 @@ pub(crate) fn process_selection_set_ir<'s>(
                     },
                 );
             }
-            FieldSelection::LocalRelationship {
+            FieldSelection::ModelRelationshipLocal {
                 query,
                 name,
                 relationship_info: _,
@@ -216,7 +237,42 @@ pub(crate) fn process_selection_set_ir<'s>(
                 }
                 ndc_fields.insert(alias.to_string(), ndc_field);
             }
-            FieldSelection::RemoteRelationship {
+            FieldSelection::CommandRelationshipLocal {
+                ir,
+                name,
+                relationship_info: _,
+            } => {
+                let (relationship_query, jl) = commands::ir_to_ndc_query(ir, join_id_counter)?;
+
+                let relationship_arguments: BTreeMap<_, _> = ir
+                    .arguments
+                    .iter()
+                    .map(|(k, v)| {
+                        (
+                            k.clone(),
+                            ndc::models::RelationshipArgument::Literal { value: v.clone() },
+                        )
+                    })
+                    .collect();
+
+                let ndc_field = ndc::models::Field::Relationship {
+                    query: Box::new(relationship_query),
+                    relationship: name.to_string(),
+                    arguments: relationship_arguments,
+                };
+
+                if !jl.locations.is_empty() {
+                    join_locations.locations.insert(
+                        alias.clone(),
+                        Location {
+                            join_node: None,
+                            rest: jl,
+                        },
+                    );
+                }
+                ndc_fields.insert(alias.to_string(), ndc_field);
+            }
+            FieldSelection::ModelRelationshipRemote {
                 ir,
                 relationship_info,
             } => {
@@ -271,20 +327,31 @@ pub(crate) fn collect_relationships(
     for field in selection.fields.values() {
         match field {
             FieldSelection::Column { .. } => (),
-            FieldSelection::LocalRelationship {
+            FieldSelection::ModelRelationshipLocal {
                 query,
                 name,
                 relationship_info,
             } => {
                 relationships.insert(
                     name.to_string(),
-                    relationship::process_relationship_definition(relationship_info)?,
+                    relationship::process_model_relationship_definition(relationship_info)?,
                 );
                 collect_relationships(&query.selection, relationships)?;
             }
+            FieldSelection::CommandRelationshipLocal {
+                ir,
+                name,
+                relationship_info,
+            } => {
+                relationships.insert(
+                    name.to_string(),
+                    relationship::process_command_relationship_definition(relationship_info)?,
+                );
+                collect_relationships(&ir.selection, relationships)?;
+            }
             // we ignore remote relationships as we are generating relationship
             // definition for one data connector
-            FieldSelection::RemoteRelationship { .. } => (),
+            FieldSelection::ModelRelationshipRemote { .. } => (),
         };
     }
     Ok(())

v3/engine/src/execute/ndc.rs

@@ -71,14 +71,14 @@ pub(crate) async fn fetch_from_data_connector<'s>(
 }
 
 /// Executes a NDC mutation
-pub(crate) async fn execute_ndc_mutation<'n, 's>(
+pub(crate) async fn execute_ndc_mutation<'n, 's, 'ir>(
     http_client: &reqwest::Client,
     query: ndc::models::MutationRequest,
     data_connector: &resolved::data_connector::DataConnector,
     selection_set: &'n normalized_ast::SelectionSet<'s, GDS>,
     execution_span_attribute: String,
     field_span_attribute: String,
-    process_response_as: ProcessResponseAs<'s>,
+    process_response_as: ProcessResponseAs<'ir>,
 ) -> Result<json::Value, error::Error> {
     let tracer = tracing_util::global_tracer();
     tracer

v3/engine/src/execute/process_response.rs

@@ -24,17 +24,24 @@ use crate::schema::{
 
 trait KeyValueResponse {
     fn remove(&mut self, key: &str) -> Option<json::Value>;
+    fn contains_key(&self, key: &str) -> bool;
 }
 impl KeyValueResponse for IndexMap<String, json::Value> {
     fn remove(&mut self, key: &str) -> Option<json::Value> {
         self.remove(key)
     }
+    fn contains_key(&self, key: &str) -> bool {
+        self.contains_key(key)
+    }
 }
 impl KeyValueResponse for IndexMap<String, RowFieldValue> {
     fn remove(&mut self, key: &str) -> Option<json::Value> {
         // Convert a RowFieldValue to json::Value if exits
         self.remove(key).map(|row_field| row_field.0)
     }
+    fn contains_key(&self, key: &str) -> bool {
+        self.contains_key(key)
+    }
 }
 
 fn process_global_id_field<T>(
@@ -115,6 +122,7 @@ where
                                             ),
                                         }
                                     })?;
+
                                 Ok(field_json_value_result)
                             }
                             OutputAnnotation::RelationshipToModel { .. } => {
@@ -152,6 +160,42 @@ where
                                     }
                                 }
                             }
+                            OutputAnnotation::RelationshipToCommand(
+                                command_relationship_annotation,
+                            ) => {
+                                let field_json_value_result =
+                                    row.remove(field.alias.0.as_str()).ok_or_else(|| {
+                                        error::InternalDeveloperError::BadGDCResponse {
+                                            summary: format!(
+                                                "missing field: {}",
+                                                field.alias.clone()
+                                            ),
+                                        }
+                                    })?;
+                                let relationship_json_value_result: Option<RowSet> =
+                                    serde_json::from_value(field_json_value_result).ok();
+
+                                match relationship_json_value_result {
+                                    None => Err(error::InternalDeveloperError::BadGDCResponse {
+                                        summary: "Unable to parse RowSet".into(),
+                                    })?,
+                                    Some(rows_set) => {
+                                        // the output of a command is optional and can be None
+                                        // so we match on the result and return null if the
+                                        // command returned None
+                                        process_command_rows(
+                                            &command_relationship_annotation.command_name,
+                                            rows_set.rows,
+                                            &field.selection_set,
+                                            &field.type_container,
+                                        )
+                                        .map(|v| match v {
+                                            None => json::Value::Null,
+                                            Some(v) => v,
+                                        })
+                                    }
+                                }
+                            }
                             _ => Err(error::InternalEngineError::UnexpectedAnnotation {
                                 annotation: annotation.clone(),
                             })?,
@@ -295,10 +339,10 @@ fn process_command_response_row(
     }
 }
 
-pub fn process_response<'s>(
+pub fn process_response(
-    selection_set: &normalized_ast::SelectionSet<'s, GDS>,
+    selection_set: &normalized_ast::SelectionSet<'_, GDS>,
     rows_sets: Vec<ndc::models::RowSet>,
-    process_response_as: ProcessResponseAs<'s>,
+    process_response_as: ProcessResponseAs,
 ) -> Result<json::Value, error::Error> {
     let tracer = tracing_util::global_tracer();
     // Post process the response to add the `__typename` fields

v3/engine/src/execute/query_plan.rs

@@ -19,11 +19,11 @@ use super::remote_joins::types::{JoinId, JoinLocations, Location, MonotonicCount
 use crate::metadata::resolved::{self, subgraph};
 use crate::schema::GDS;
 
-pub type QueryPlan<'n, 's> = IndexMap<ast::Alias, NodeQueryPlan<'n, 's>>;
+pub type QueryPlan<'n, 's, 'ir> = IndexMap<ast::Alias, NodeQueryPlan<'n, 's, 'ir>>;
 
 /// Query plan of individual root field or node
 #[derive(Debug)]
-pub enum NodeQueryPlan<'n, 's> {
+pub enum NodeQueryPlan<'n, 's, 'ir> {
     /// __typename field on query root
     TypeName { type_name: ast::TypeName },
     /// __schema field
@@ -40,30 +40,33 @@ pub enum NodeQueryPlan<'n, 's> {
         role: Role,
     },
     /// NDC query to be executed
-    NDCQueryExecution(NDCQueryExecution<'n, 's>),
+    NDCQueryExecution(NDCQueryExecution<'s, 'ir>),
     /// NDC query for Relay 'node' to be executed
-    RelayNodeSelect(Option<NDCQueryExecution<'n, 's>>),
+    RelayNodeSelect(Option<NDCQueryExecution<'s, 'ir>>),
     /// NDC mutation to be executed
-    NDCMutationExecution(NDCMutationExecution<'n, 's>),
+    NDCMutationExecution(NDCMutationExecution<'n, 's, 'ir>),
 }
 
 #[derive(Debug)]
-pub struct NDCQueryExecution<'n, 's> {
+pub struct NDCQueryExecution<'s, 'ir> {
     pub execution_tree: ExecutionTree<'s>,
     pub execution_span_attribute: String,
     pub field_span_attribute: String,
-    pub process_response_as: ProcessResponseAs<'s>,
+    pub process_response_as: ProcessResponseAs<'ir>,
-    pub selection_set: &'n normalized_ast::SelectionSet<'s, GDS>,
+    // This selection set can either be owned by the IR structures or by the normalized query request itself.
+    // We use the more restrictive lifetime `'ir` here which allows us to construct this struct using the selection
+    // set either from the IR or from the normalized query request.
+    pub selection_set: &'ir normalized_ast::SelectionSet<'s, GDS>,
 }
 
 #[derive(Debug)]
-pub struct NDCMutationExecution<'n, 's> {
+pub struct NDCMutationExecution<'n, 's, 'ir> {
     pub query: ndc_client::models::MutationRequest,
     pub join_locations: JoinLocations<(RemoteJoin<'s>, JoinId)>,
     pub data_connector: &'s resolved::data_connector::DataConnector,
     pub execution_span_attribute: String,
     pub field_span_attribute: String,
-    pub process_response_as: ProcessResponseAs<'s>,
+    pub process_response_as: ProcessResponseAs<'ir>,
     pub selection_set: &'n normalized_ast::SelectionSet<'s, GDS>,
 }
 
@@ -80,7 +83,7 @@ pub struct ExecutionNode<'s> {
 }
 
 #[derive(Debug)]
-pub enum ProcessResponseAs<'s> {
+pub enum ProcessResponseAs<'ir> {
     Object {
         is_nullable: bool,
     },
@@ -88,12 +91,12 @@ pub enum ProcessResponseAs<'s> {
         is_nullable: bool,
     },
     CommandResponse {
-        command_name: &'s subgraph::Qualified<open_dds::commands::CommandName>,
+        command_name: &'ir subgraph::Qualified<open_dds::commands::CommandName>,
-        type_container: &'s ast::TypeContainer<ast::TypeName>,
+        type_container: &'ir ast::TypeContainer<ast::TypeName>,
     },
 }
 
-impl ProcessResponseAs<'_> {
+impl<'ir> ProcessResponseAs<'ir> {
     pub fn is_nullable(&self) -> bool {
         match self {
             ProcessResponseAs::Object { is_nullable } => *is_nullable,
@@ -103,9 +106,9 @@ impl ProcessResponseAs<'_> {
     }
 }
 
-pub fn generate_query_plan<'n, 's>(
+pub fn generate_query_plan<'n, 's, 'ir>(
-    ir: &'s IndexMap<ast::Alias, root_field::RootField<'n, 's>>,
+    ir: &'ir IndexMap<ast::Alias, root_field::RootField<'n, 's>>,
-) -> Result<QueryPlan<'n, 's>, error::Error> {
+) -> Result<QueryPlan<'n, 's, 'ir>, error::Error> {
     let mut query_plan = IndexMap::new();
     for (alias, field) in ir.into_iter() {
         let field_plan = match field {
@@ -117,9 +120,9 @@ pub fn generate_query_plan<'n, 's>(
     Ok(query_plan)
 }
 
-fn plan_mutation<'n, 's>(
+fn plan_mutation<'n, 's, 'ir>(
-    ir: &'s root_field::MutationRootField<'n, 's>,
+    ir: &'ir root_field::MutationRootField<'n, 's>,
-) -> Result<NodeQueryPlan<'n, 's>, error::Error> {
+) -> Result<NodeQueryPlan<'n, 's, 'ir>, error::Error> {
     let plan = match ir {
         root_field::MutationRootField::TypeName { type_name } => NodeQueryPlan::TypeName {
             type_name: type_name.clone(),
@@ -141,13 +144,13 @@ fn plan_mutation<'n, 's>(
             NodeQueryPlan::NDCMutationExecution(NDCMutationExecution {
                 query: ndc_ir,
                 join_locations: join_locations_ids,
-                data_connector: &ir.data_connector,
+                data_connector: ir.data_connector,
                 selection_set,
                 execution_span_attribute: "execute_command".into(),
                 field_span_attribute: ir.field_name.to_string(),
                 process_response_as: ProcessResponseAs::CommandResponse {
                     command_name: &ir.command_name,
-                    type_container: ir.type_container,
+                    type_container: &ir.type_container,
                 },
             })
         }
@@ -155,9 +158,9 @@ fn plan_mutation<'n, 's>(
     Ok(plan)
 }
 
-fn plan_query<'n, 's>(
+fn plan_query<'n, 's, 'ir>(
-    ir: &'s root_field::QueryRootField<'n, 's>,
+    ir: &'ir root_field::QueryRootField<'n, 's>,
-) -> Result<NodeQueryPlan<'n, 's>, error::Error> {
+) -> Result<NodeQueryPlan<'n, 's, 'ir>, error::Error> {
     let mut counter = MonotonicCounter::new();
     let query_plan = match ir {
         root_field::QueryRootField::TypeName { type_name } => NodeQueryPlan::TypeName {
@@ -238,7 +241,7 @@ fn plan_query<'n, 's>(
             let execution_tree = ExecutionTree {
                 root_node: ExecutionNode {
                     query: ndc_ir,
-                    data_connector: &ir.data_connector,
+                    data_connector: ir.data_connector,
                 },
                 remote_executions: join_locations_ids,
             };
@@ -249,7 +252,7 @@ fn plan_query<'n, 's>(
                 field_span_attribute: ir.field_name.to_string(),
                 process_response_as: ProcessResponseAs::CommandResponse {
                     command_name: &ir.command_name,
-                    type_container: ir.type_container,
+                    type_container: &ir.type_container,
                 },
             })
         }
@@ -257,7 +260,7 @@ fn plan_query<'n, 's>(
     Ok(query_plan)
 }
 
-fn generate_execution_tree<'s>(ir: &'s ModelSelection) -> Result<ExecutionTree<'s>, error::Error> {
+fn generate_execution_tree<'s>(ir: &ModelSelection<'s>) -> Result<ExecutionTree<'s>, error::Error> {
     let mut counter = MonotonicCounter::new();
     let (ndc_ir, join_locations) = model_selection::ir_to_ndc_ir(ir, &mut counter)?;
     let join_locations_with_ids = assign_with_join_ids(join_locations)?;
@@ -413,9 +416,9 @@ impl ExecuteQueryResult {
     }
 }
 
-pub async fn execute_query_plan<'n, 's>(
+pub async fn execute_query_plan<'n, 's, 'ir>(
     http_client: &reqwest::Client,
-    query_plan: QueryPlan<'n, 's>,
+    query_plan: QueryPlan<'n, 's, 'ir>,
 ) -> ExecuteQueryResult {
     let mut root_fields = IndexMap::new();
     for (alias, field_plan) in query_plan.into_iter() {
@@ -540,7 +543,7 @@ async fn resolve_ndc_query_execution(
 
 async fn resolve_ndc_mutation_execution(
     http_client: &reqwest::Client,
-    ndc_query: NDCMutationExecution<'_, '_>,
+    ndc_query: NDCMutationExecution<'_, '_, '_>,
 ) -> Result<json::Value, error::Error> {
     let NDCMutationExecution {
         query,

v3/engine/src/execute/remote_joins.rs

@@ -74,7 +74,7 @@ pub async fn execute_join_locations(
     execution_span_attribute: String,
     field_span_attribute: String,
     lhs_response: &mut Vec<ndc::models::RowSet>,
-    lhs_response_type: &ProcessResponseAs<'_>,
+    lhs_response_type: &ProcessResponseAs,
     join_locations: JoinLocations<(RemoteJoin<'async_recursion>, JoinId)>,
 ) -> Result<(), error::Error> {
     let tracer = tracing_util::global_tracer();
@@ -177,7 +177,7 @@ struct CollectArgumentResult<'s> {
 /// a structure of `ReplacementToken`s.
 fn collect_arguments<'s>(
     lhs_response: &Vec<ndc::models::RowSet>,
-    lhs_response_type: &ProcessResponseAs<'s>,
+    lhs_response_type: &ProcessResponseAs,
     key: &str,
     location: &Location<(RemoteJoin<'s>, JoinId)>,
     arguments: &mut Arguments,

v3/engine/src/schema.rs

@@ -193,7 +193,9 @@ pub enum Error {
         type_name: ast::TypeName,
     },
     #[error("internal error while building schema, command not found: {command_name}")]
-    InternalCommandNotFound { command_name: CommandName },
+    InternalCommandNotFound {
+        command_name: Qualified<CommandName>,
+    },
     #[error("Cannot generate select_many API for model {model_name} since order_by_expression isn't defined")]
     NoOrderByExpression { model_name: Qualified<ModelName> },
     #[error("No graphql type name has been defined for scalar type: {type_name}")]

v3/engine/src/schema/commands.rs

@@ -7,6 +7,7 @@ use lang_graphql::schema as gql_schema;
 use lang_graphql::schema::InputField;
 use lang_graphql::schema::Namespaced;
 use ndc_client as gdc;
+use open_dds::arguments::ArgumentName;
 use std::collections::HashMap;
 
 use crate::metadata::resolved;
@@ -23,6 +24,40 @@ pub enum Response {
     },
 }
 
+pub(crate) fn generate_command_argument(
+    gds: &GDS,
+    builder: &mut gql_schema::Builder<GDS>,
+    command: &resolved::command::Command,
+    argument_name: &ArgumentName,
+    argument_type: &crate::schema::commands::resolved::subgraph::QualifiedTypeReference,
+) -> Result<(ast::Name, Namespaced<GDS, InputField<GDS>>), crate::schema::Error> {
+    let field_name = ast::Name::new(argument_name.0.as_str())?;
+    let input_type = types::input_type::get_input_type(gds, builder, argument_type)?;
+    Ok((
+        field_name.clone(),
+        builder.allow_all_namespaced(
+            gql_schema::InputField::new(
+                field_name,
+                None,
+                Annotation::Input(types::InputAnnotation::CommandArgument {
+                    argument_type: argument_type.clone(),
+                    ndc_func_proc_argument: command
+                        .source
+                        .as_ref()
+                        .and_then(|command_source| {
+                            command_source.argument_mappings.get(argument_name)
+                        })
+                        .cloned(),
+                }),
+                input_type,
+                None,
+                gql_schema::DeprecationStatus::NotDeprecated,
+            ),
+            None,
+        ),
+    ))
+}
+
 pub(crate) fn command_field(
     gds: &GDS,
     builder: &mut gql_schema::Builder<GDS>,
@@ -39,28 +74,8 @@ pub(crate) fn command_field(
 
     let mut arguments = HashMap::new();
     for (argument_name, argument_type) in &command.arguments {
-        let field_name = ast::Name::new(argument_name.0.as_str())?;
+        let (field_name, input_field) =
-        let input_type = types::input_type::get_input_type(gds, builder, argument_type)?;
+            generate_command_argument(gds, builder, command, argument_name, argument_type)?;
-        let input_field: Namespaced<GDS, InputField<GDS>> = builder.allow_all_namespaced(
-            gql_schema::InputField::new(
-                field_name.clone(),
-                None,
-                Annotation::Input(types::InputAnnotation::CommandArgument {
-                    argument_type: argument_type.clone(),
-                    ndc_func_proc_argument: command
-                        .source
-                        .as_ref()
-                        .and_then(|command_source| {
-                            command_source.argument_mappings.get(argument_name)
-                        })
-                        .cloned(),
-                }),
-                input_type,
-                None,
-                gql_schema::DeprecationStatus::NotDeprecated,
-            ),
-            None,
-        );
         arguments.insert(field_name, input_field);
     }
 

v3/engine/src/schema/permissions.rs

@@ -50,7 +50,7 @@ pub(crate) fn get_select_one_namespace_annotations(
         .collect()
 }
 
-/// Build namespace annotation for relationship permissions.
+/// Build namespace annotation for model relationship permissions.
 /// We need to check the permissions of the source and target fields
 /// in the relationship mappings.
 pub(crate) fn get_model_relationship_namespace_annotations(
@@ -95,6 +95,30 @@ pub(crate) fn get_command_namespace_annotations(
     permissions
 }
 
+/// Build namespace annotation for command relationship permissions.
+/// We need to check the permissions of the source fields
+/// in the relationship mappings.
+pub(crate) fn get_command_relationship_namespace_annotations(
+    command: &resolved::command::Command,
+    source_object_type_representation: &ObjectTypeRepresentation,
+    mappings: &[resolved::relationship::RelationshipCommandMapping],
+) -> HashMap<Role, Option<types::NamespaceAnnotation>> {
+    let select_permissions = get_command_namespace_annotations(command);
+
+    select_permissions
+        .into_iter()
+        .filter(|(role, _)| {
+            mappings.iter().all(|mapping| {
+                get_allowed_roles_for_field(
+                    source_object_type_representation,
+                    &mapping.source_field.field_name,
+                )
+                .any(|allowed_role| role == allowed_role)
+            })
+        })
+        .collect()
+}
+
 /// Build namespace annotations for the node interface..
 /// The global ID field and the Node interface will only be exposed
 /// for a role if the role has access (select permissions)

v3/engine/src/schema/types.rs

@@ -111,6 +111,7 @@ pub enum OutputAnnotation {
         global_id_fields: Vec<types::FieldName>,
     },
     RelationshipToModel(output_type::relationship::ModelRelationshipAnnotation),
+    RelationshipToCommand(output_type::relationship::CommandRelationshipAnnotation),
     RelayNodeInterfaceID {
         typename_mappings: HashMap<ast::TypeName, Vec<types::FieldName>>,
     },

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

@@ -1,12 +1,15 @@
 use lang_graphql::ast::common as ast;
-use lang_graphql::schema as gql_schema;
+use lang_graphql::schema::{self as gql_schema};
 use open_dds::{
     relationships,
     types::{CustomTypeName, InbuiltType},
 };
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
-use self::relationship::{ModelRelationshipAnnotation, ModelTargetSource};
+use self::relationship::{
+    CommandRelationshipAnnotation, CommandTargetSource, ModelRelationshipAnnotation,
+    ModelTargetSource,
+};
 use super::inbuilt_type::base_type_container_for_inbuilt_type;
 use super::Annotation;
 use crate::metadata::resolved::subgraph::{
@@ -17,6 +20,7 @@ use crate::metadata::resolved::{
     self,
     types::{mk_name, TypeRepresentation},
 };
+use crate::schema::commands::generate_command_argument;
 use crate::schema::permissions;
 use crate::schema::query_root::select_many::generate_select_many_arguments;
 use crate::schema::{Role, GDS};
@@ -167,10 +171,70 @@ fn object_type_fields(
                 let graphql_field_name = relationship_field_name.clone();
 
                 let relationship_field = match &relationship.target {
-                    resolved::relationship::RelationshipTarget::Command { .. } => {
+                    resolved::relationship::RelationshipTarget::Command {
-                        return Err(Error::InternalUnsupported {
+                        command_name,
-                            summary: "Relationships to commands aren't supported".into(),
+                        target_type,
-                        });
+                        mappings,
+                    } => {
+                        let relationship_output_type = get_output_type(gds, builder, target_type)?;
+
+                        let command = gds.metadata.commands.get(command_name).ok_or_else(|| {
+                            Error::InternalCommandNotFound {
+                                command_name: command_name.clone(),
+                            }
+                        })?;
+
+                        let mut arguments_with_mapping = HashSet::new();
+                        for argument_mapping in mappings {
+                            arguments_with_mapping.insert(&argument_mapping.argument_name);
+                        }
+
+                        // generate argument fields for the command arguments which are not mapped to
+                        // any type fields, so that they can be exposed in the relationship field schema
+                        let mut arguments = HashMap::new();
+                        for (argument_name, argument_type) in &command.arguments {
+                            if !arguments_with_mapping.contains(argument_name) {
+                                let (field_name, input_field) = generate_command_argument(
+                                    gds,
+                                    builder,
+                                    command,
+                                    argument_name,
+                                    argument_type,
+                                )?;
+                                arguments.insert(field_name, input_field);
+                            }
+                        }
+
+                        builder.conditional_namespaced(
+                            gql_schema::Field::<GDS>::new(
+                                graphql_field_name.clone(),
+                                None,
+                                Annotation::Output(super::OutputAnnotation::RelationshipToCommand(
+                                    CommandRelationshipAnnotation {
+                                        source_type: relationship.source.clone(),
+                                        relationship_name: relationship.name.clone(),
+                                        command_name: command_name.clone(),
+                                        target_source: CommandTargetSource::new(
+                                            command,
+                                            relationship,
+                                        )?,
+                                        target_type: target_type.clone(),
+                                        underlying_object_typename: command
+                                            .underlying_object_typename
+                                            .clone(),
+                                        mappings: mappings.clone(),
+                                    },
+                                )),
+                                relationship_output_type,
+                                arguments,
+                                gql_schema::DeprecationStatus::NotDeprecated,
+                            ),
+                            permissions::get_command_relationship_namespace_annotations(
+                                command,
+                                object_type_representation,
+                                mappings,
+                            ),
+                        )
                     }
                     resolved::relationship::RelationshipTarget::Model {
                         model_name,

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

@@ -1,4 +1,5 @@
 use open_dds::{
+    commands::CommandName,
     models::ModelName,
     relationships::{RelationshipName, RelationshipType},
     types::CustomTypeName,
@@ -7,7 +8,10 @@ use open_dds::{
 use serde::{Deserialize, Serialize};
 
 use crate::{
-    metadata::resolved::{self, subgraph::Qualified},
+    metadata::resolved::{
+        self,
+        subgraph::{Qualified, QualifiedTypeReference},
+    },
     schema,
 };
 
@@ -52,3 +56,45 @@ impl ModelTargetSource {
             .transpose()
     }
 }
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+pub struct CommandRelationshipAnnotation {
+    pub source_type: Qualified<CustomTypeName>,
+    pub relationship_name: RelationshipName,
+    pub command_name: Qualified<CommandName>,
+    pub target_source: Option<CommandTargetSource>,
+    pub target_type: QualifiedTypeReference,
+    pub underlying_object_typename: Option<Qualified<CustomTypeName>>,
+    pub mappings: Vec<resolved::relationship::RelationshipCommandMapping>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+pub struct CommandTargetSource {
+    pub(crate) command: resolved::command::CommandSource,
+    pub(crate) capabilities: resolved::relationship::RelationshipCapabilities,
+}
+
+impl CommandTargetSource {
+    pub fn new(
+        command: &resolved::command::Command,
+        relationship: &resolved::relationship::Relationship,
+    ) -> Result<Option<Self>, schema::Error> {
+        command
+            .source
+            .as_ref()
+            .map(|command_source| {
+                Ok(Self {
+                    command: command_source.clone(),
+                    capabilities: relationship
+                        .target_capabilities
+                        .as_ref()
+                        .ok_or_else(|| schema::Error::InternalMissingRelationshipCapabilities {
+                            type_name: relationship.source.clone(),
+                            relationship: relationship.name.clone(),
+                        })?
+                        .clone(),
+                })
+            })
+            .transpose()
+    }
+}

v3/engine/tests/execute/common_metadata/command_metadata.json

@@ -48,9 +48,32 @@
                 "typeName": "CommandActor"
               }
             }
+          },
+          {
+            "kind": "ObjectType",
+            "version": "v1",
+            "definition": {
+              "name": "commandMovie",
+              "fields": [
+                {
+                  "name": "movie_id",
+                  "type": "Int!"
+                },
+                {
+                  "name": "title",
+                  "type": "String!"
+                },
+                {
+                  "name": "rating",
+                  "type": "Int!"
+                }
+              ],
+              "graphql": {
+                "typeName": "CommandMovie"
+              }
+            }
           }
         ]
       }
     ]
   }
-  

v3/engine/tests/execute/common_metadata/custom_connector_schema.json

@@ -253,6 +253,26 @@
                     }
                   }
                 },
+                {
+                  "name": "get_movie_by_id",
+                  "description": "Get movie by ID",
+                  "arguments": {
+                    "id": {
+                      "description": "the id of the movie to fetch",
+                      "type": {
+                        "type": "named",
+                        "name": "Int"
+                      }
+                    }
+                  },
+                  "result_type": {
+                    "type": "nullable",
+                    "underlying_type": {
+                      "type": "named",
+                      "name": "movie"
+                    }
+                  }
+                },
                 {
                   "name": "get_actors_by_name",
                   "description": "Get actors by name",

v3/engine/tests/execute/relationships/model_to_command/expected.json

@@ -0,0 +1,72 @@
+[
+  {
+    "data": {
+      "ActorMany": [
+        {
+          "MovieFromCommand": {
+            "movie_id": 1,
+            "rating": 4,
+            "title": "Titanic"
+          },
+          "name": "Leonardo DiCaprio"
+        },
+        {
+          "MovieFromCommand": {
+            "movie_id": 1,
+            "rating": 4,
+            "title": "Titanic"
+          },
+          "name": "Kate Winslet"
+        },
+        {
+          "MovieFromCommand": {
+            "movie_id": 2,
+            "rating": 5,
+            "title": "Slumdog Millionaire"
+          },
+          "name": "Irfan Khan"
+        },
+        {
+          "MovieFromCommand": {
+            "movie_id": 3,
+            "rating": 4,
+            "title": "Godfather"
+          },
+          "name": "Al Pacino"
+        },
+        {
+          "MovieFromCommand": {
+            "movie_id": 3,
+            "rating": 4,
+            "title": "Godfather"
+          },
+          "name": "Robert De Niro"
+        }
+      ]
+    }
+  },
+  {
+    "data": null,
+    "errors": [
+      {
+        "message": "validation failed: field rating on type CommandMovie is not allowed for user"
+      }
+    ]
+  },
+  {
+    "data": null,
+    "errors": [
+      {
+        "message": "validation failed: field MovieFromCommand on type Actor is not allowed for user_without_perm_on_movie_id"
+      }
+    ]
+  },
+  {
+    "data": null,
+    "errors": [
+      {
+        "message": "validation failed: field MovieFromCommand on type Actor is not allowed for user_without_perm_on_command"
+      }
+    ]
+  }
+]

v3/engine/tests/execute/relationships/model_to_command/metadata.json

@@ -0,0 +1,332 @@
+{
+  "version": "v2",
+  "subgraphs": [
+    {
+      "name": "default",
+      "objects": [
+        {
+          "kind": "ObjectType",
+          "version": "v1",
+          "definition": {
+            "name": "actor",
+            "fields": [
+              {
+                "name": "actor_id",
+                "type": "Int!"
+              },
+              {
+                "name": "name",
+                "type": "String!"
+              },
+              {
+                "name": "movie_id",
+                "type": "Int!"
+              }
+            ],
+            "graphql": {
+              "typeName": "Actor"
+            }
+          }
+        },
+        {
+          "kind": "TypePermissions",
+          "version": "v1",
+          "definition": {
+            "typeName": "actor",
+            "permissions": [
+              {
+                "role": "admin",
+                "output": {
+                  "allowedFields": [
+                    "actor_id",
+                    "name",
+                    "movie_id"
+                  ]
+                }
+              },
+              {
+                "role": "user",
+                "output": {
+                  "allowedFields": [
+                    "actor_id",
+                    "name",
+                    "movie_id"
+                  ]
+                }
+              },
+              {
+                "role": "user_without_perm_on_movie_id",
+                "output": {
+                  "allowedFields": [
+                    "actor_id",
+                    "name"
+                  ]
+                }
+              },
+              {
+                "role": "user_without_perm_on_command",
+                "output": {
+                  "allowedFields": [
+                    "actor_id",
+                    "name",
+                    "movie_id"
+                  ]
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "Model",
+          "version": "v1",
+          "definition": {
+            "name": "Actors",
+            "objectType": "actor",
+            "source": {
+              "dataConnectorName": "custom",
+              "collection": "actors",
+              "typeMapping": {
+                "actor": {
+                  "fieldMapping": {
+                    "actor_id": {
+                      "column": "id"
+                    },
+                    "name": {
+                      "column": "name"
+                    },
+                    "movie_id": {
+                      "column": "movie_id"
+                    }
+                  }
+                }
+              }
+            },
+            "graphql": {
+              "selectUniques": [],
+              "selectMany": {
+                "queryRootField": "ActorMany"
+              }
+            },
+            "filterableFields": [
+              {
+                "fieldName": "actor_id",
+                "operators": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "name",
+                "operators": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "movie_id",
+                "operators": {
+                  "enableAll": true
+                }
+              }
+            ],
+            "orderableFields": [
+              {
+                "fieldName": "actor_id",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "name",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              },
+              {
+                "fieldName": "movie_id",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "ModelPermissions",
+          "version": "v1",
+          "definition": {
+              "modelName": "Actors",
+              "permissions": [
+                  {
+                      "role": "admin",
+                      "select": {
+                          "filter": null
+                      }
+                  },
+                  {
+                    "role": "user",
+                    "select": {
+                        "filter": null
+                    }
+                  },
+                  {
+                    "role": "user_without_perm_on_movie_id",
+                    "select": {
+                        "filter": null
+                    }
+                  },
+                  {
+                    "role": "user_without_perm_on_command",
+                    "select": {
+                        "filter": null
+                    }
+                  }
+              ]
+          }
+        },
+        {
+          "kind": "TypePermissions",
+          "version": "v1",
+          "definition": {
+            "typeName": "commandMovie",
+            "permissions": [
+              {
+                "role": "admin",
+                "output": {
+                  "allowedFields": [
+                    "movie_id",
+                    "title",
+                    "rating"
+                  ]
+                }
+              },
+              {
+                "role": "user",
+                "output": {
+                  "allowedFields": [
+                    "movie_id",
+                    "title"
+                  ]
+                }
+              },
+              {
+                "role": "user_without_perm_on_movie_id",
+                "output": {
+                  "allowedFields": [
+                    "movie_id",
+                    "title",
+                    "rating"
+                  ]
+                }
+              },
+              {
+                "role": "user_without_perm_on_command",
+                "output": {
+                  "allowedFields": [
+                    "movie_id",
+                    "title",
+                    "rating"
+                  ]
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "CommandPermissions",
+          "version": "v1",
+          "definition": {
+            "commandName": "get_movie_by_id",
+            "permissions": [
+              {
+                "role": "admin",
+                "allowExecution": true
+              },
+              {
+                "role": "user",
+                "allowExecution": true
+              },
+              {
+                "role": "user_without_perm_on_movie_id",
+                "allowExecution": true
+              },
+              {
+                "role": "user_without_perm_on_command",
+                "allowExecution": false
+              }
+            ]
+          }
+        },
+        {
+          "kind": "Command",
+          "version": "v1",
+          "definition": {
+            "name": "get_movie_by_id",
+            "arguments": [
+              {
+                "name": "movie_id",
+                "type": "Int!"
+              }
+            ],
+            "outputType": "commandMovie",
+            "source": {
+              "dataConnectorName": "custom",
+              "dataConnectorCommand": {
+                "function": "get_movie_by_id"
+              },
+              "typeMapping": {
+                "commandMovie": {
+                  "fieldMapping": {
+                    "movie_id": {
+                      "column": "id"
+                    },
+                    "title": {
+                      "column": "title"
+                    },
+                    "rating": {
+                      "column": "rating"
+                    }
+                  }
+                }
+              },
+              "argumentMapping": {
+                "movie_id": "id"
+              }
+            },
+            "graphql": {
+              "rootFieldName": "getMovieById",
+              "rootFieldKind": "Query"
+            }
+          }
+        },
+        {
+          "definition": {
+              "source": "actor",
+              "name": "MovieFromCommand",
+              "target": {
+                  "command": {
+                      "name": "get_movie_by_id"
+                  }
+              },
+              "mapping": [
+                  {
+                      "source": {
+                          "fieldPath": [
+                              {
+                                  "fieldName": "movie_id"
+                              }
+                          ]
+                      },
+                      "target": {
+                          "argument": {
+                              "argumentName": "movie_id"
+                          }
+                      }
+                  }
+              ]
+          },
+          "version": "v1",
+          "kind": "Relationship"
+        }
+      ]
+    }
+  ]
+}

v3/engine/tests/execute/relationships/model_to_command/request.gql

@@ -0,0 +1,10 @@
+query MyQuery {
+  ActorMany {
+    MovieFromCommand {
+      movie_id
+      rating
+      title
+    }
+    name
+  }
+}

v3/engine/tests/execute/relationships/model_to_command/session_variables.json

@@ -0,0 +1,14 @@
+[
+    {
+        "x-hasura-role": "admin"
+    },
+    {
+        "x-hasura-role": "user"
+    },
+    {
+        "x-hasura-role": "user_without_perm_on_movie_id"
+    },
+    {
+        "x-hasura-role": "user_without_perm_on_command"
+    }
+]

v3/engine/tests/generate_ir/get_many_model_count/expected.json

@@ -435,7 +435,7 @@
           "selection": {
             "fields": {
               "article": {
-                "LocalRelationship": {
+                "ModelRelationshipLocal": {
                   "query": {
                     "data_connector": {
                       "name": {
@@ -458,7 +458,7 @@
                     "selection": {
                       "fields": {
                         "Author": {
-                          "LocalRelationship": {
+                          "ModelRelationshipLocal": {
                             "query": {
                               "data_connector": {
                                 "name": {
@@ -481,7 +481,7 @@
                               "selection": {
                                 "fields": {
                                   "Articles": {
-                                    "LocalRelationship": {
+                                    "ModelRelationshipLocal": {
                                       "query": {
                                         "data_connector": {
                                           "name": {

v3/engine/tests/relationship.rs

@@ -53,6 +53,20 @@ fn test_local_relationships_command_to_model() {
     common::test_execution_expectation(test_path_string, &[common_metadata_path_string]);
 }
 
+#[test]
+fn test_local_relationships_model_to_command() {
+    let test_path_string = "execute/relationships/model_to_command";
+    let common_command_metadata_path_string = "execute/common_metadata/command_metadata.json";
+    let common_metadata_path_string = "execute/common_metadata/custom_connector_schema.json";
+    common::test_execution_expectation(
+        test_path_string,
+        &[
+            common_metadata_path_string,
+            common_command_metadata_path_string,
+        ],
+    );
+}
+
 #[test]
 fn test_local_relationships_permissions_target_model_filter_predicate() {
     let test_path_string = "execute/relationships/permissions/target_model_filter_predicate";

v3/engine/tests/schema.json

@@ -630,7 +630,7 @@
                 },
                 {
                   "name": "get_article_by_id",
-                  "description": "Insert or update an article",
+                  "description": "Get article based on ID",
                   "arguments": {
                     "id": {
                       "description": "the id of the article to fetch",
@@ -647,6 +647,26 @@
                       "name": "article"
                     }
                   }
+                },
+                {
+                  "name": "get_author_by_id",
+                  "description": "Get artist based on id",
+                  "arguments": {
+                    "id": {
+                      "description": "the id of the artist to fetch",
+                      "type": {
+                        "type": "named",
+                        "name": "int4"
+                      }
+                    }
+                  },
+                  "result_type": {
+                    "type": "nullable",
+                    "underlying_type": {
+                      "type": "named",
+                      "name": "author"
+                    }
+                  }
                 }
               ],
               "procedures": [
@@ -886,6 +906,59 @@
             ]
           }
         },
+        {
+          "kind": "ObjectType",
+          "version": "v1",
+          "definition": {
+            "name": "commandAuthor",
+            "fields": [
+              {
+                "name": "id",
+                "type": "Int!"
+              },
+              {
+                "name": "first_name",
+                "type": "String!"
+              },
+              {
+                "name": "last_name",
+                "type": "String!"
+              }
+            ],
+            "graphql": {
+              "typeName": "CommandAuthor"
+            }
+          }
+        },
+        {
+          "kind": "TypePermissions",
+          "version": "v1",
+          "definition": {
+            "typeName": "commandAuthor",
+            "permissions": [
+              {
+                "role": "admin",
+                "output": {
+                  "allowedFields": [
+                    "id",
+                    "first_name",
+                    "last_name"
+                  ]
+                }
+              },
+              {
+                "role": "user",
+                "output": {
+                  "allowedFields": [
+                    "id",
+                    "first_name",
+                    "last_name"
+                  ]
+                }
+              }
+            ]
+          }
+        },
         {
           "kind": "CommandPermissions",
           "version": "v1",
@@ -945,6 +1018,66 @@
             }
           }
         },
+
+        {
+          "kind": "CommandPermissions",
+          "version": "v1",
+          "definition": {
+            "commandName": "get_author_by_id",
+            "permissions": [
+              {
+                "role": "admin",
+                "allowExecution": true
+              },
+              {
+                "role": "user",
+                "allowExecution": true
+              }
+            ]
+          }
+        },
+        {
+          "kind": "Command",
+          "version": "v1",
+          "definition": {
+            "name": "get_author_by_id",
+            "arguments": [
+              {
+                "name": "author_id",
+                "type": "Int!"
+              }
+            ],
+            "outputType": "commandAuthor",
+            "source": {
+              "dataConnectorName": "db",
+              "dataConnectorCommand": {
+                "function": "get_author_by_id"
+              },
+              "typeMapping": {
+                "commandArticle": {
+                  "fieldMapping": {
+                    "author_id": {
+                      "column": "id"
+                    },
+                    "first_name": {
+                      "column": "first_name"
+                    },
+                    "last_name": {
+                      "column": "last_name"
+                    }
+                  }
+                }
+              },
+              "argumentMapping": {
+                "author_id": "id"
+              }
+            },
+            "graphql": {
+              "rootFieldName": "getAuthorById",
+              "rootFieldKind": "Query"
+            }
+          }
+        },
         {
           "kind": "CommandPermissions",
           "version": "v1",
@@ -2444,6 +2577,35 @@
           "version": "v1",
           "kind": "Relationship"
         },
+        {
+          "definition": {
+            "source": "article",
+            "name": "AuthorFromCommand",
+            "target": {
+              "command": {
+                "name": "get_author_by_id"
+              }
+            },
+            "mapping": [
+              {
+                "source": {
+                  "fieldPath": [
+                    {
+                      "fieldName": "author_id"
+                    }
+                  ]
+                },
+                "target": {
+                  "argument": {
+                    "argumentName": "author_id"
+                  }
+                }
+              }
+            ]
+          },
+          "version": "v1",
+          "kind": "Relationship"
+        },
         {
           "definition": {
             "source": "Track",