Diff execution plans in tests (#1688)

<!-- The PR description should answer 2 important questions: -->

### What

All the tests were passing, but since we're going to diff execution
plans in prod, need to remove false positives, so this PR changes our
engine tests to diff execution plans between old and new pipelines.

We fix everything to make these pass. Next step is to actually run these
in prod and emit traces for failures.

V3_GIT_ORIGIN_REV_ID: 289e07016b6210d052736697ca3c71760ebade40

Author: danieljharvey

v3/Cargo.lock

@@ -138,6 +138,7 @@ serde_json = { version = "1", features = ["preserve_order"] }
 serde_path_to_error = "0.1"
 serde_with = { version = "3", features = ["indexmap_2"] }
 sha2 = "0.10"
+similar-asserts = "1.7"
 smol_str = "0.1"
 strum = "0.26"
 strum_macros = "0.26"

v3/Cargo.toml

@@ -70,6 +70,7 @@ graphql-ir = { path = "../graphql/ir" }
 criterion = { workspace = true }
 goldenfile = { workspace = true }
 pretty_assertions = { workspace = true }
+similar-asserts = { workspace = true }
 tokio-test = { workspace = true }
 
 [package.metadata.cargo-machete]

v3/crates/engine/Cargo.toml

@@ -402,6 +402,15 @@ pub fn test_execution_expectation_for_multiple_ndc_versions(
                             "OpenDD pipeline",
                         );
 
+                        // check execution plans are the same
+                        diff_execution_plan(
+                            &schema,
+                            &arc_resolved_metadata.clone(),
+                            session,
+                            &request_headers,
+                            &raw_request,
+                        );
+
                         responses.push(http_response);
                     }
                 }
@@ -479,6 +488,16 @@ pub fn test_execution_expectation_for_multiple_ndc_versions(
                             &open_dd_response.inner(),
                             "OpenDD pipeline",
                         );
+
+                        // check execution plans are the same
+                        diff_execution_plan(
+                            &schema,
+                            &arc_resolved_metadata.clone(),
+                            session,
+                            &request_headers,
+                            &raw_request,
+                        );
+
                         responses.push(http_response);
                     }
                 }
@@ -770,3 +789,58 @@ async fn run_query_graphql_ws(
     }
     response
 }
+
+/// Compare executions plans. If they fail to build, ignore them.
+pub fn diff_execution_plan(
+    schema: &'_ lang_graphql::schema::Schema<GDS>,
+    metadata: &'_ Arc<metadata_resolve::Metadata>,
+    session: &Session,
+    request_headers: &reqwest::header::HeaderMap,
+    raw_request: &lang_graphql::http::RawRequest,
+) {
+    // parse the raw request into a GQL query
+    if let Ok(query) = graphql_frontend::parse_query(&raw_request.query) {
+        // normalize the parsed GQL query
+        if let Ok(normalized_request) =
+            graphql_frontend::normalize_request(schema, session, query, raw_request)
+        {
+            // generate IR
+            if let Ok(old_ir) = graphql_frontend::build_ir(
+                GraphqlRequestPipeline::Old,
+                schema,
+                metadata,
+                session,
+                request_headers,
+                &normalized_request,
+            ) {
+                // construct a plan to execute the request
+                if let Ok(old_request_plan) = graphql_frontend::build_request_plan(
+                    &old_ir,
+                    metadata,
+                    session,
+                    request_headers,
+                ) {
+                    // generate IR
+                    if let Ok(new_ir) = graphql_frontend::build_ir(
+                        GraphqlRequestPipeline::OpenDd,
+                        schema,
+                        metadata,
+                        session,
+                        request_headers,
+                        &normalized_request,
+                    ) {
+                        // construct a plan to execute the request
+                        if let Ok(new_request_plan) = graphql_frontend::build_request_plan(
+                            &new_ir,
+                            metadata,
+                            session,
+                            request_headers,
+                        ) {
+                            similar_asserts::assert_eq!(old_request_plan, new_request_plan);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

v3/crates/engine/tests/common.rs

@@ -34,18 +34,6 @@ pub(crate) fn plan_filter_expression(
         );
     }
 
-    if let Some(filter) = relationship_join_filter {
-        expressions.push(
-            plan_expression(filter, relationships, &mut remote_predicates, unique_number).map_err(
-                |plan_error| {
-                    plan_error::Error::Internal(plan_error::InternalError::InternalGeneric {
-                        description: plan_error.to_string(),
-                    })
-                },
-            )?,
-        );
-    }
-
     if let Some(filter) = &query_filter.additional_filter {
         expressions.push(
             plan_expression(filter, relationships, &mut remote_predicates, unique_number).map_err(
@@ -69,8 +57,20 @@ pub(crate) fn plan_filter_expression(
         expressions.push(planned_expression);
     }
 
-    Ok((
-        ResolvedFilterExpression::mk_and(expressions).remove_always_true_expression(),
-        remote_predicates,
-    ))
+    if let Some(filter) = relationship_join_filter {
+        expressions.push(
+            plan_expression(filter, relationships, &mut remote_predicates, unique_number).map_err(
+                |plan_error| {
+                    plan_error::Error::Internal(plan_error::InternalError::InternalGeneric {
+                        description: plan_error.to_string(),
+                    })
+                },
+            )?,
+        );
+    }
+
+    let resolved_filter_expression =
+        ResolvedFilterExpression::mk_and(expressions).remove_always_true_expression();
+
+    Ok((resolved_filter_expression, remote_predicates))
 }

v3/crates/graphql/ir/src/plan/filter.rs

@@ -17,7 +17,7 @@ use plan_types::{
     SourceFieldAlias, TargetField,
 };
 use plan_types::{NdcFieldAlias, NdcRelationshipName, Relationship, UniqueNumber};
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
 
 pub(crate) fn plan_nested_selection(
     nested_selection: &NestedSelection<'_>,
@@ -256,7 +256,7 @@ pub(crate) fn plan_selection_set(
                 ir,
                 relationship_info,
             } => {
-                let mut join_mapping = HashMap::new();
+                let mut join_mapping = BTreeMap::new();
                 for ((src_field_alias, src_field), target_field) in &relationship_info.join_mapping
                 {
                     let ndc_field_alias = process_remote_relationship_field_mapping(
@@ -304,7 +304,7 @@ pub(crate) fn plan_selection_set(
                 ir,
                 relationship_info,
             } => {
-                let mut join_mapping = HashMap::new();
+                let mut join_mapping = BTreeMap::new();
 
                 for ((src_field_alias, src_field), target_field) in &relationship_info.join_mapping
                 {

v3/crates/graphql/ir/src/plan/selection_set.rs

@@ -851,9 +851,12 @@ pub fn build_remote_relationship<'s>(
         }
     }
 
-    remote_relationships_ir
-        .filter_clause
-        .relationship_join_filter = Some(Expression::mk_and(relationship_join_filter_expressions));
+    if !relationship_join_filter_expressions.is_empty() {
+        remote_relationships_ir
+            .filter_clause
+            .relationship_join_filter =
+            Some(Expression::mk_and(relationship_join_filter_expressions));
+    }
     remote_relationships_ir.variable_arguments = variable_arguments;
 
     let rel_info = RemoteModelRelationshipInfo { join_mapping };

v3/crates/graphql/ir/src/relationship.rs

@@ -54,19 +54,21 @@ impl ResolvedFilterExpression {
         if expressions.len() == 1 {
             expressions.into_iter().next().unwrap()
         }
-        // If all subexpressions are also `and`, we can flatten into a single `and`
-        // ie. and([and([x,y]), and([a,b])]) == and([x,y,a,b])
+        // If any subexpressions are also `and`, we can flatten into a single `and`
+        // ie. and([and([x,y]), and([a,b])]) == and([x,y,a,b]) or
+        // and([a, and([b,c])]) == and([a,b,c])
         else if expressions
             .iter()
-            .all(|expr| matches!(expr, ResolvedFilterExpression::And { .. }))
+            .any(|expr| matches!(expr, ResolvedFilterExpression::And { .. }))
         {
             let subexprs = expressions
                 .into_iter()
                 .flat_map(|expr| match expr {
                     ResolvedFilterExpression::And { expressions } => expressions,
-                    _ => vec![],
+                    other => vec![other],
                 })
                 .collect();
+
             ResolvedFilterExpression::And {
                 expressions: subexprs,
             }

v3/crates/plan-types/src/execution_plan/filter.rs

@@ -7,7 +7,6 @@ use open_dds::arguments::ArgumentName;
 use open_dds::types::FieldName;
 use serde::Serialize;
 use std::collections::BTreeMap;
-use std::collections::HashMap;
 use std::sync::Arc;
 
 /// This tree structure captures all the locations (in the selection set IR) where
@@ -115,15 +114,15 @@ pub struct RemoteJoin {
     /// NDC node to execute on a data connector
     pub target_ndc_execution: query::QueryExecutionPlan,
     /// Mapping of the fields in source to fields in target.
-    /// The HashMap has the following info -
+    /// The BTreeMap has the following info -
     ///   - key: is the field name in the source
     ///   - value->first item: is the alias we create for the
     ///     source field. If the user did not request the join field in the
     ///     selection set, we include the join mapping field and call it a phantom
     ///     field.
     ///   - value->second item: is the target NDC field. This could be a model
     ///     field or an argument name.
-    pub join_mapping: HashMap<SourceFieldName, (SourceFieldAlias, TargetField)>,
+    pub join_mapping: BTreeMap<SourceFieldName, (SourceFieldAlias, TargetField)>,
     /// Represents how to process the join response.
     pub process_response_as: ProcessResponseAs,
     /// Represents the type of the remote join

v3/crates/plan-types/src/execution_plan/remote_joins.rs

@@ -82,19 +82,21 @@ impl<'s> Expression<'s> {
         if expressions.len() == 1 {
             expressions.into_iter().next().unwrap()
         }
-        // If all subexpressions are also `and`, we can flatten into a single `and`
+        // If any subexpressions are also `and`, we can flatten into a single `and`
         // ie. and([and([x,y]), and([a,b])]) == and([x,y,a,b])
+        // ie. and([a, and([b,c])]) == and([a,b,c])
         else if expressions
             .iter()
-            .all(|expr| matches!(expr, Expression::And { .. }))
+            .any(|expr| matches!(expr, Expression::And { .. }))
         {
             let subexprs = expressions
                 .into_iter()
                 .flat_map(|expr| match expr {
                     Expression::And { expressions } => expressions,
-                    _ => vec![],
+                    other => vec![other],
                 })
                 .collect();
+
             Expression::And {
                 expressions: subexprs,
             }