Bring ndc-postgres in line with ndc-spec RC15

crates/configuration/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 [dependencies]
 query-engine-metadata = { path = "../query-engine/metadata" }
 
-ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "02d26c1" }
+ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "e0e9629" }
 
 schemars = { version = "0.8.16", features = ["smol_str", "preserve_order"] }
 serde = "1.0.196"

crates/connectors/ndc-postgres/Cargo.toml

@@ -21,7 +21,7 @@ query-engine-metadata = { path = "../../query-engine/metadata" }
 query-engine-sql = { path = "../../query-engine/sql" }
 query-engine-translation = { path = "../../query-engine/translation" }
 
-ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "02d26c1" }
+ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "e0e9629" }
 
 async-trait = "0.1.77"
 percent-encoding = "2.3.1"
@@ -34,8 +34,8 @@ tracing = "0.1.40"
 url = "2.5.0"
 
 [dev-dependencies]
-ndc-client = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.0-rc.14" }
-ndc-test = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.0-rc.14" }
+ndc-client = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.0-rc.15" }
+ndc-test = { git = "https://github.com/hasura/ndc-spec.git", tag = "v0.1.0-rc.15" }
 tests-common = { path = "../../tests/tests-common" }
 
 axum = "0.6.20"

crates/connectors/ndc-postgres/src/schema.rs

@@ -21,7 +21,7 @@ pub async fn get_schema(
 ) -> Result<models::SchemaResponse, connector::SchemaError> {
     let configuration::RuntimeConfiguration { metadata, .. } = config;
 
-    let scalar_types: BTreeMap<String, models::ScalarType> =
+    let mut scalar_types: BTreeMap<String, models::ScalarType> =
         configuration::occurring_scalar_types(&metadata.tables, &metadata.native_queries)
             .iter()
             .map(|scalar_type| {
@@ -260,7 +260,9 @@ pub async fn get_schema(
             config.mutations_version,
         )
         .iter()
-        .map(|(name, mutation)| mutation_to_procedure(name, mutation, &mut more_object_types))
+        .map(|(name, mutation)| {
+            mutation_to_procedure(name, mutation, &mut more_object_types, &mut scalar_types)
+        })
         .collect();
 
     procedures.extend(generated_procedures);
@@ -301,17 +303,25 @@ fn mutation_to_procedure(
     name: &String,
     mutation: &generate::Mutation,
     object_types: &mut BTreeMap<String, models::ObjectType>,
+    scalar_types: &mut BTreeMap<String, models::ScalarType>,
 ) -> models::ProcedureInfo {
     match mutation {
-        generate::Mutation::DeleteMutation(delete) => delete_to_procedure(name, delete),
+        generate::Mutation::DeleteMutation(delete) => {
+            delete_to_procedure(name, delete, object_types, scalar_types)
+        }
         generate::Mutation::InsertMutation(insert) => {
-            insert_to_procedure(name, insert, object_types)
+            insert_to_procedure(name, insert, object_types, scalar_types)
         }
     }
 }
 
 /// given a `DeleteMutation`, turn it into a `ProcedureInfo` to be output in the schema
-fn delete_to_procedure(name: &String, delete: &delete::DeleteMutation) -> models::ProcedureInfo {
+fn delete_to_procedure(
+    name: &String,
+    delete: &delete::DeleteMutation,
+    object_types: &mut BTreeMap<String, models::ObjectType>,
+    scalar_types: &mut BTreeMap<String, models::ScalarType>,
+) -> models::ProcedureInfo {
     match delete {
         delete::DeleteMutation::DeleteByKey {
             by_column,
@@ -329,14 +339,16 @@ fn delete_to_procedure(name: &String, delete: &delete::DeleteMutation) -> models
                 },
             );
 
-            models::ProcedureInfo {
-                name: name.to_string(),
-                description: Some(description.to_string()),
+            make_procedure_type(
+                name.to_string(),
+                Some(description.to_string()),
                 arguments,
-                result_type: models::Type::Named {
+                models::Type::Named {
                     name: collection_name.to_string(),
                 },
-            }
+                object_types,
+                scalar_types,
+            )
         }
     }
 }
@@ -379,6 +391,7 @@ fn insert_to_procedure(
     name: &String,
     insert: &insert::InsertMutation,
     object_types: &mut BTreeMap<String, models::ObjectType>,
+    scalar_types: &mut BTreeMap<String, models::ScalarType>,
 ) -> models::ProcedureInfo {
     let mut arguments = BTreeMap::new();
     let object_type = make_object_type(&insert.columns);
@@ -393,12 +406,83 @@ fn insert_to_procedure(
         },
     );
 
+    make_procedure_type(
+        name.to_string(),
+        Some(insert.description.to_string()),
+        arguments,
+        models::Type::Named {
+            name: insert.collection_name.to_string(),
+        },
+        object_types,
+        scalar_types,
+    )
+}
+
+/// Build a `ProcedureInfo` type from the given parameters.
+///
+/// Because procedures return an `affected_rows` count alongside the result type that it's
+/// `returning`, we have to generate a separate object type for its result. As part of that, we may
+/// also have to include the `int4` scalar type (if it isn't included for another reason elsewhere
+/// in the schema). So, this function creates that object type, optionally adds that scalar type,
+/// and then returns a `ProcedureInfo` that points to the correct object type.
+fn make_procedure_type(
+    name: String,
+    description: Option<String>,
+    arguments: BTreeMap<String, models::ArgumentInfo>,
+    result_type: models::Type,
+
+    object_types: &mut BTreeMap<String, models::ObjectType>,
+    scalar_types: &mut BTreeMap<String, models::ScalarType>,
+) -> models::ProcedureInfo {
+    let mut fields = BTreeMap::new();
+    let object_type_name = format!("{name}_response");
+
+    // If int4 doesn't exist anywhere else in the schema, we need to add it here. However, a user
+    // can't filter or aggregate based on the affected rows of a procedure, so we don't need to add
+    // any aggregate functions or comparison operators. However, if int4 exists elsewhere in the
+    // schema and has already been added, it will also already contain these functions and
+    // operators.
+    scalar_types
+        .entry("int4".to_string())
+        .or_insert(models::ScalarType {
+            aggregate_functions: BTreeMap::new(),
+            comparison_operators: BTreeMap::new(),
+        });
+
+    fields.insert(
+        "affected_rows".to_string(),
+        models::ObjectField {
+            description: Some("The number of rows affected by the mutation".to_string()),
+            r#type: models::Type::Named {
+                name: "int4".to_string(),
+            },
+        },
+    );
+
+    fields.insert(
+        "returning".to_string(),
+        models::ObjectField {
+            description: Some("Data from rows affected by the mutation".to_string()),
+            r#type: models::Type::Array {
+                element_type: Box::from(result_type),
+            },
+        },
+    );
+
+    object_types.insert(
+        object_type_name.clone(),
+        models::ObjectType {
+            description: Some(format!("Responses from the '{name}' procedure")),
+            fields,
+        },
+    );
+
     models::ProcedureInfo {
-        name: name.to_string(),
-        description: Some(insert.description.to_string()),
+        name,
+        description,
         arguments,
         result_type: models::Type::Named {
-            name: insert.collection_name.to_string(),
+            name: object_type_name,
         },
     }
 }

crates/query-engine/sql/src/sql/helpers.rs

@@ -382,7 +382,8 @@ pub fn select_rowset_with_variables(
     final_select
 }
 
-/// Given a set of rows and aggregate queries, combine them into one Select.
+/// Build a `Select` query using a `SelectSet` of row fields and aggregates according to the
+/// following SQL template:
 ///
 /// ```sql
 /// SELECT row_to_json(<output_table_alias>) AS <output_column_alias>
@@ -397,16 +398,29 @@ pub fn select_rowset_with_variables(
 ///     ) AS <aggregate_table_alias>
 /// ) AS <output_table_alias>
 /// ```
+///
+/// The `SelectSet` determines whether we select from both the rows and the aggregates, or just the
+/// rows, or just the aggregates.
 pub fn select_mutation_rowset(
     (output_table_alias, output_column_alias): (TableAlias, ColumnAlias),
     (row_table_alias, row_column_alias): (TableAlias, ColumnAlias),
     aggregate_table_alias: TableAlias,
-    row_select: Select,
-    aggregate_select: Option<Select>,
+    select: SelectSet,
 ) -> Select {
     let row = vec![(
         output_column_alias,
-        Expression::RowToJson(TableReference::AliasedTable(output_table_alias.clone())),
+        Expression::JsonBuildObject(BTreeMap::from([
+            (
+                "type".to_string(),
+                Box::new(Expression::Value(Value::String("procedure".to_string()))),
+            ),
+            (
+                "result".to_string(),
+                Box::new(Expression::RowToJson(TableReference::AliasedTable(
+                    output_table_alias.clone(),
+                ))),
+            ),
+        ])),
     )];
 
     let mut final_select = simple_select(row);
@@ -415,25 +429,43 @@ pub fn select_mutation_rowset(
         select_rows_as_json_for_mutation(row_sel, row_column_alias, row_table_alias.clone())
     };
 
-    let mut select_star = star_select(From::Select {
-        alias: row_table_alias.clone(),
-        select: Box::new(wrap_row(row_select)),
-    });
+    match select {
+        SelectSet::Rows(row_select) => {
+            let select_star = star_select(From::Select {
+                alias: row_table_alias.clone(),
+                select: Box::new(wrap_row(row_select)),
+            });
+
+            final_select.from = Some(From::Select {
+                alias: output_table_alias,
+                select: Box::new(select_star),
+            });
+        }
+
+        SelectSet::Aggregates(aggregate_select) => {
+            final_select.from = Some(From::Select {
+                alias: output_table_alias,
+                select: Box::new(aggregate_select),
+            });
+        }
+
+        SelectSet::RowsAndAggregates(row_select, aggregate_select) => {
+            let mut select_star = star_select(From::Select {
+                alias: row_table_alias.clone(),
+                select: Box::new(wrap_row(row_select)),
+            });
 
-    match aggregate_select {
-        None => {}
-        Some(aggregate_select) => {
             select_star.joins = vec![Join::CrossJoin(CrossJoin {
                 select: Box::new(aggregate_select),
                 alias: aggregate_table_alias.clone(),
             })];
-        }
-    }
 
-    final_select.from = Some(From::Select {
-        alias: output_table_alias,
-        select: Box::new(select_star),
-    });
+            final_select.from = Some(From::Select {
+                alias: output_table_alias,
+                select: Box::new(select_star),
+            });
+        }
+    };
 
     final_select
 }
@@ -505,14 +537,7 @@ pub fn select_rows_as_json_for_mutation(
             Expression::Value(Value::EmptyJsonArray),
         ],
     };
-    let wrapped_expression = Expression::FunctionCall {
-        function: Function::JsonBuildArray,
-        args: vec![Expression::JsonBuildObject(BTreeMap::from([(
-            "__value".to_string(),
-            Box::new(expression),
-        )]))],
-    };
-    let mut select = simple_select(vec![(column_alias, wrapped_expression)]);
+    let mut select = simple_select(vec![(column_alias, expression)]);
     select.from = Some(From::Select {
         select: Box::new(row_select),
         alias: table_alias,

crates/query-engine/translation/Cargo.toml

@@ -5,7 +5,7 @@ edition = "2021"
 license = "Apache-2.0"
 
 [dependencies]
-ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "02d26c1" }
+ndc-sdk = { git = "https://github.com/hasura/ndc-hub.git", rev = "e0e9629" }
 query-engine-metadata = { path = "../metadata" }
 query-engine-sql = { path = "../sql" }
 

crates/query-engine/translation/src/translation/error.rs

@@ -28,6 +28,8 @@ pub enum Error {
     ColumnIsIdentityAlways(String),
     MissingColumnInInsert(String, String),
     NotImplementedYet(String),
+    NoProcedureResultFieldsRequested,
+    UnexpectedStructure(String),
     InternalError(String),
 }
 
@@ -119,6 +121,11 @@ impl std::fmt::Display for Error {
             Error::NotImplementedYet(thing) => {
                 write!(f, "Queries containing {} are not supported.", thing)
             }
+            Error::NoProcedureResultFieldsRequested => write!(
+                f,
+                "Procedure requests must ask for 'affected_rows' or use the 'returning' clause."
+            ),
+            Error::UnexpectedStructure(structure) => write!(f, "Unexpected {}.", structure),
             Error::InternalError(thing) => {
                 write!(f, "Internal error: {}.", thing)
             }