fix(router, executor): fix abstract type project and improve projection plan performance (#633)

#641 +
#634 + the following:

This PR introduces three related improvements to debugging the
projection logic:
- pretty-printing of projection plans
- a new `projection` command in `dev` CLI to pretty-print the plan for a
given operation and supergrah
- refactor to reduce condition to their minimal forms when possible

The new `dev` command works like the `plan` command, you pass both
supergraph and operation and get back a pretty-printed response
projection plan. This helped me to understand why ROUTER-235 issue
occured.

The simplification of field projection conditions was needed to make the
pretty-printed plan much cleaner and easier to understand, but also to
make the checking more efficient at runtime (when the response
projection happens - now there should be fewer conditions to check).

An example:
```
$ cargo dev projection supergraph.graphql op.graphql

contentPage: {
  conditions: (ParentType(Exact(Query)) AND FieldType(Exact(ContentPage)))
  selections:
    contentBody: {
      conditions: (ParentType(Exact(ContentPage)) AND FieldType(Exact(ContentContainer)))
      selections:
        section: {
          conditions: (ParentType(Exact(ContentContainer)) AND FieldType(OneOf(IContent, ContentA, ContentB)))
          selections:
            contentChildren: {
              conditions: ((ParentType(Exact(ContentA)) AND FieldType(Exact(ContentAChild))) OR (ParentType(Exact(ContentB)) AND FieldType(Exact(ContentBChild))))
              selections:
                title: {
                  conditions: ((ParentType(Exact(ContentAChild)) AND FieldType(Exact(String))) OR (ParentType(Exact(ContentBChild)) AND FieldType(Exact(String))))
                }
            }
        }
    }
}
```

---------

Co-authored-by: Dotan Simha <dotansimha@gmail.com>

Author: kamilkisiela

.changeset/projection-fix.md

@@ -0,0 +1,14 @@
+---
+router: patch
+executor: patch
+---
+
+Fixed response projection for fields on different concrete types of interfaces and unions.
+
+When a query included fragments on an abstract type (interface or union) that selected fields with the same name but different return types, the projection would incorrectly use a single, merged plan for all types. This caused projection to fail when processing responses where different concrete types had different field implementations.
+
+For example, with `... on A { children { id } }` and `... on B { children { id } }` where `A.children` returns `[AChild]` and `B.children` returns `[BChild]`, the projection would fail to correctly distinguish between the types and return incomplete or incorrect data.
+
+The fix introduces type-aware plan merging, which preserves the context of which concrete types a field came from. During response projection, the type is now determined dynamically for each object, ensuring the correct field type is used.
+
+In addition, a refactor of the response projection logic was performed to improve performance.

Cargo.lock

@@ -9,6 +9,7 @@ path = "src/main.rs"
 
 [dependencies]
 hive-router-query-planner = { path = "../../lib/query-planner" }
+hive-router-plan-executor = { path = "../../lib/executor" }
 
 graphql-parser = { workspace = true }
 tracing-subscriber = { workspace = true }

bin/dev-cli/Cargo.toml

@@ -1,6 +1,8 @@
 use std::env;
 use std::process;
 
+use hive_router_plan_executor::introspection::schema::SchemaWithMetadata;
+use hive_router_plan_executor::projection::plan::FieldProjectionPlan;
 use hive_router_query_planner::ast::normalization::normalize_operation;
 use hive_router_query_planner::ast::operation::OperationDefinition;
 use hive_router_query_planner::consumer_schema::ConsumerSchema;
@@ -101,6 +103,26 @@ fn main() {
 
             println!("{}", operation);
         }
+        "projection" => {
+            let supergraph_sdl =
+                std::fs::read_to_string(&args[2]).expect("Unable to read input file");
+            let parsed_schema = parse_schema(&supergraph_sdl);
+            let supergraph = SupergraphState::new(&parsed_schema);
+            let document_text =
+                std::fs::read_to_string(&args[3]).expect("Unable to read input file");
+            let parsed_document = parse_operation(&document_text);
+            let document = normalize_operation(&supergraph, &parsed_document, None).unwrap();
+            let operation = document.executable_operation();
+            let consumer_schema = ConsumerSchema::new_from_supergraph(&parsed_schema);
+            let schema_metadata = consumer_schema.schema_metadata();
+
+            let (_, projection_plan) =
+                FieldProjectionPlan::from_operation(operation, &schema_metadata);
+
+            for plan in &projection_plan {
+                println!("{}", plan);
+            }
+        }
         _ => {
             eprintln!("Unknown command. Available commands: consumer_graph, graph, paths, tree, fetch_graph, plan");
             process::exit(1);

bin/dev-cli/src/main.rs

@@ -24,10 +24,9 @@ fn project_data_by_operation_test(c: &mut Criterion) {
     let normalized_document = normalize_operation(&planner.supergraph, &parsed_document, None)
         .expect("Failed to normalize operation");
     let normalized_operation = normalized_document.executable_operation();
-    let (root_type_name, projection_plan) = FieldProjectionPlan::from_operation(
-        normalized_operation,
-        &planner.consumer_schema.schema_metadata(),
-    );
+    let schema_metadata = &planner.consumer_schema.schema_metadata();
+    let (root_type_name, projection_plan) =
+        FieldProjectionPlan::from_operation(normalized_operation, schema_metadata);
     let result_as_string = raw_result::get_result_as_string();
     let projected_data_as_json: sonic_rs::Value =
         sonic_rs::from_slice(result_as_string.as_bytes()).unwrap();
@@ -48,11 +47,12 @@ fn project_data_by_operation_test(c: &mut Criterion) {
                     &bb_projection_plan,
                     &None,
                     result_as_string.len(),
+                    schema_metadata,
                 )
                 .unwrap();
                 black_box(result);
             },
-            criterion::BatchSize::LargeInput,
+            criterion::BatchSize::SmallInput,
         );
     });
 }

lib/executor/benches/executor_benches.rs

@@ -8,7 +8,7 @@ use crate::{
 
 #[derive(thiserror::Error, Debug, Clone, IntoStaticStr)]
 pub enum PlanExecutionErrorKind {
-    #[error("Projection faiure: {0}")]
+    #[error("Projection failure: {0}")]
     #[strum(serialize = "PROJECTION_FAILURE")]
     ProjectionFailure(#[from] ProjectionError),
 

lib/executor/src/execution/error.rs

@@ -114,6 +114,7 @@ pub async fn execute_query_plan<'exec, 'req>(
         ctx.projection_plan,
         ctx.variable_values,
         exec_ctx.response_storage.estimate_final_response_size(),
+        ctx.introspection_context.metadata,
     )
     .with_plan_context(LazyPlanContext {
         subgraph_name: || None,

lib/executor/src/execution/plan.rs

@@ -21,6 +21,7 @@ pub struct SchemaMetadata {
     pub object_types: HashSet<String>,
     pub scalar_types: HashSet<String>,
     pub union_types: HashSet<String>,
+    pub interface_types: HashSet<String>,
 }
 
 impl SchemaMetadata {
@@ -36,6 +37,10 @@ impl SchemaMetadata {
         self.union_types.contains(name)
     }
 
+    pub fn is_interface_type(&self, name: &str) -> bool {
+        self.interface_types.contains(name)
+    }
+
     pub fn get_type_fields(&self, type_name: &str) -> Option<&HashMap<String, FieldTypeInfo>> {
         self.type_fields.get(type_name)
     }
@@ -88,6 +93,7 @@ impl SchemaWithMetadata for ConsumerSchema {
         ]);
         let mut object_types: HashSet<String> = HashSet::default();
         let mut union_types: HashSet<String> = HashSet::default();
+        let mut interface_types: HashSet<String> = HashSet::default();
 
         for definition in &self.document.definitions {
             match definition {
@@ -123,6 +129,7 @@ impl SchemaWithMetadata for ConsumerSchema {
                 }
                 Definition::TypeDefinition(TypeDefinition::Interface(interface_type)) => {
                     let name = interface_type.name.to_string();
+                    interface_types.insert(name.clone());
                     let mut fields = HashMap::default();
                     for field in &interface_type.fields {
                         let field_type_name = field.field_type.type_name();
@@ -183,6 +190,7 @@ impl SchemaWithMetadata for ConsumerSchema {
             object_types,
             scalar_types,
             union_types,
+            interface_types,
         }
     }
 }

lib/executor/src/introspection/schema.rs

@@ -6,4 +6,11 @@ pub enum ProjectionError {
     ExtensionsSerializationFailure(String),
     #[error("Failed to serialize a custom scalar: {0}")]
     CustomScalarSerializationFailure(String),
+    #[error("Type '{0}' not found in schema")]
+    MissingType(String),
+    #[error("Field '{field_name}' not found on type '{type_name}' in schema")]
+    MissingField {
+        field_name: String,
+        type_name: String,
+    },
 }