GraphQL Mutation support for datawarehouse. (#1978)

## Why make this change?

Issue: #1976. 

GraphQL Mutations need to be supported for data warehouse. Reasons for
listed changes listed in issue document.
This pr does not cover REST addition for insert/patch etc. Will be done
in following pr's . issue tracking:
#1982

## What is this change?

1. Adding a DbOperationResult type that will be the return type for all
dw mutation operations.
2. Generating DWSQL queries for create, update, delete and upsert
queries similar to that of other data sources.
3. Adding model directive to DW mutation nodes so that we can determine
which db they belong to.

## How was this tested?
Performed integration tests:

![image](https://github.com/Azure/data-api-builder/assets/124841904/59d896a6-b5c3-41d8-9fc6-f6a277dcfe81)

- [x] Unit Tests
Unit tests added for all dw scenarios.

src/Core/Resolvers/DWSqlQueryBuilder.cs

@@ -15,6 +15,7 @@ namespace Azure.DataApiBuilder.Core.Resolvers
     public class DwSqlQueryBuilder : BaseSqlQueryBuilder, IQueryBuilder
     {
         private static DbCommandBuilder _builder = new SqlCommandBuilder();
+        public const string COUNT_ROWS_WITH_GIVEN_PK = "cnt_rows_to_update";
 
         /// <inheritdoc />
         public override string QuoteIdentifier(string ident)
@@ -161,19 +162,48 @@ private static string GenerateColumnsAsJson(SqlQueryStructure structure, bool su
         /// <inheritdoc />
         public string Build(SqlInsertStructure structure)
         {
-            throw new NotImplementedException("DataWarehouse Sql currently does not support inserts");
+            string tableName = $"{QuoteIdentifier(structure.DatabaseObject.SchemaName)}.{QuoteIdentifier(structure.DatabaseObject.Name)}";
+
+            // Predicates by virtue of database policy for Create action.
+            string dbPolicypredicates = JoinPredicateStrings(structure.GetDbPolicyForOperation(EntityActionOperation.Create));
+
+            // Columns whose values are provided in the request body - to be inserted into the record.
+            string insertColumns = Build(structure.InsertColumns);
+
+            // Values to be inserted into the entity.
+            string values = dbPolicypredicates.Equals(BASE_PREDICATE) ?
+                $"VALUES ({string.Join(", ", structure.Values)});" : $"SELECT {insertColumns} FROM (VALUES({string.Join(", ", structure.Values)})) T({insertColumns}) WHERE {dbPolicypredicates};";
+
+            // Final insert query to be executed against the database.
+            StringBuilder insertQuery = new();
+            insertQuery.Append($"INSERT INTO {tableName} ({insertColumns}) ");
+            insertQuery.Append(values);
+
+            return insertQuery.ToString();
         }
 
         /// <inheritdoc />
         public string Build(SqlUpdateStructure structure)
         {
-            throw new NotImplementedException("DataWarehouse sql currently does not support updates");
+            string tableName = $"{QuoteIdentifier(structure.DatabaseObject.SchemaName)}.{QuoteIdentifier(structure.DatabaseObject.Name)}";
+            string predicates = JoinPredicateStrings(
+                       structure.GetDbPolicyForOperation(EntityActionOperation.Update),
+                       Build(structure.Predicates));
+
+            StringBuilder updateQuery = new($"UPDATE {tableName} SET {Build(structure.UpdateOperations, ", ")} ");
+            updateQuery.Append($"WHERE {predicates};");
+            return updateQuery.ToString();
         }
 
         /// <inheritdoc />
         public string Build(SqlDeleteStructure structure)
         {
-            throw new NotImplementedException("DataWarehouse sql currently does not support deletes");
+            string predicates = JoinPredicateStrings(
+                       structure.GetDbPolicyForOperation(EntityActionOperation.Delete),
+                       Build(structure.Predicates));
+
+            return $"DELETE FROM {QuoteIdentifier(structure.DatabaseObject.SchemaName)}.{QuoteIdentifier(structure.DatabaseObject.Name)} " +
+                    $"WHERE {predicates} ";
         }
 
         /// <inheritdoc />
@@ -185,7 +215,65 @@ public string Build(SqlExecuteStructure structure)
         /// <inheritdoc />
         public string Build(SqlUpsertQueryStructure structure)
         {
-            throw new NotImplementedException("DataWarehouse sql currently does not support updates");
+            string tableName = $"{QuoteIdentifier(structure.DatabaseObject.SchemaName)}.{QuoteIdentifier(structure.DatabaseObject.Name)}";
+
+            // Predicates by virtue of PK.
+            string pkPredicates = JoinPredicateStrings(Build(structure.Predicates));
+
+            string updateOperations = Build(structure.UpdateOperations, ", ");
+            string queryToGetCountOfRecordWithPK = $"SELECT COUNT(*) as {COUNT_ROWS_WITH_GIVEN_PK} FROM {tableName} WHERE {pkPredicates}";
+
+            // Query to get the number of records with a given PK.
+            string prefixQuery = $"DECLARE @ROWS_TO_UPDATE int;" +
+                $"SET @ROWS_TO_UPDATE = ({queryToGetCountOfRecordWithPK}); " +
+                $"{queryToGetCountOfRecordWithPK};";
+
+            // Final query to be executed for the given PUT/PATCH operation.
+            StringBuilder upsertQuery = new(prefixQuery);
+
+            // Query to update record (if there exists one for given PK).
+            StringBuilder updateQuery = new(
+                $"IF @ROWS_TO_UPDATE = 1 " +
+                $"BEGIN " +
+                $"UPDATE {tableName} " +
+                $"SET {updateOperations} ");
+
+            // End the IF block.
+            updateQuery.Append("END ");
+
+            // Append the update query to upsert query.
+            upsertQuery.Append(updateQuery);
+            if (!structure.IsFallbackToUpdate)
+            {
+                // Append the conditional to check if the insert query is to be executed or not.
+                // Insert is only attempted when no record exists corresponding to given PK.
+                upsertQuery.Append("ELSE BEGIN ");
+
+                // Columns which are assigned some value in the PUT/PATCH request.
+                string insertColumns = Build(structure.InsertColumns);
+
+                // Predicates added by virtue of database policy for create operation.
+                string createPredicates = JoinPredicateStrings(structure.GetDbPolicyForOperation(EntityActionOperation.Create));
+
+                // Query to insert record (if there exists none for given PK).
+                StringBuilder insertQuery = new($"INSERT INTO {tableName} ({insertColumns}) ");
+
+                // Query to fetch the column values to be inserted into the entity.
+                string fetchColumnValuesQuery = BASE_PREDICATE.Equals(createPredicates) ?
+                    $"VALUES({string.Join(", ", structure.Values)});" :
+                    $"SELECT {insertColumns} FROM (VALUES({string.Join(", ", structure.Values)})) T({insertColumns}) WHERE {createPredicates};";
+
+                // Append the values to be inserted to the insertQuery.
+                insertQuery.Append(fetchColumnValuesQuery);
+
+                // Append the insert query to the upsert query.
+                upsertQuery.Append(insertQuery.ToString());
+
+                // End the ELSE block.
+                upsertQuery.Append("END");
+            }
+
+            return upsertQuery.ToString();
         }
 
         /// <summary>

src/Core/Resolvers/Factories/MutationEngineFactory.cs

@@ -51,6 +51,7 @@ public class MutationEngineFactory : IMutationEngineFactory
                 _mutationEngines.Add(DatabaseType.MySQL, mutationEngine);
                 _mutationEngines.Add(DatabaseType.MSSQL, mutationEngine);
                 _mutationEngines.Add(DatabaseType.PostgreSQL, mutationEngine);
+                _mutationEngines.Add(DatabaseType.DWSQL, mutationEngine);
             }
 
             if (config.CosmosDataSourceUsed)

src/Core/Resolvers/Sql Query Structures/SqlQueryStructure.cs

@@ -302,7 +302,7 @@ private List<OrderByColumn> PrimaryKeyAsOrderByColumns()
                 PaginationMetadata.Subqueries.Add(QueryBuilder.PAGINATION_FIELD_NAME, PaginationMetadata.MakeEmptyPaginationMetadata());
             }
 
-            EntityName = _underlyingFieldType.Name;
+            EntityName = sqlMetadataProvider.GetDatabaseType() == DatabaseType.DWSQL ? GraphQLUtils.GetEntityNameFromContext(ctx) : _underlyingFieldType.Name;
 
             if (GraphQLUtils.TryExtractGraphQLFieldModelName(_underlyingFieldType.Directives, out string? modelName))
             {

src/Core/Resolvers/SqlMutationEngine.cs

@@ -83,14 +83,7 @@ public class SqlMutationEngine : IMutationEngine
 
             dataSourceName = GetValidatedDataSourceName(dataSourceName);
             string graphqlMutationName = context.Selection.Field.Name.Value;
-            IOutputType outputType = context.Selection.Field.Type;
-            string entityName = outputType.TypeName();
-            ObjectType _underlyingFieldType = GraphQLUtils.UnderlyingGraphQLEntityType(outputType);
-
-            if (GraphQLUtils.TryExtractGraphQLFieldModelName(_underlyingFieldType.Directives, out string? modelName))
-            {
-                entityName = modelName;
-            }
+            string entityName = GraphQLUtils.GetEntityNameFromContext(context);
 
             ISqlMetadataProvider sqlMetadataProvider = _sqlMetadataProviderFactory.GetMetadataProvider(dataSourceName);
             IQueryEngine queryEngine = _queryEngineFactory.GetQueryEngine(sqlMetadataProvider.GetDatabaseType());
@@ -119,12 +112,17 @@ public class SqlMutationEngine : IMutationEngine
                         // be computed only when the read permission is configured.
                         if (isReadPermissionConfigured)
                         {
-                            // compute the mutation result before removing the element,
-                            // since typical GraphQL delete mutations return the metadata of the deleted item.
-                            result = await queryEngine.ExecuteAsync(
-                                        context,
-                                        GetBackingColumnsFromCollection(entityName: entityName, parameters: parameters, sqlMetadataProvider: sqlMetadataProvider),
-                                        dataSourceName);
+                            // For cases we only require a result summarizing the operation (DBOperationResult),
+                            // we can skip getting the impacted records.
+                            if (context.Selection.Type.TypeName() != GraphQLUtils.DB_OPERATION_RESULT_TYPE)
+                            {
+                                // compute the mutation result before removing the element,
+                                // since typical GraphQL delete mutations return the metadata of the deleted item.
+                                result = await queryEngine.ExecuteAsync(
+                                            context,
+                                            GetBackingColumnsFromCollection(entityName: entityName, parameters: parameters, sqlMetadataProvider: sqlMetadataProvider),
+                                            dataSourceName);
+                            }
                         }
 
                         Dictionary<string, object>? resultProperties =
@@ -134,16 +132,28 @@ public class SqlMutationEngine : IMutationEngine
                                 sqlMetadataProvider);
 
                         // If the number of records affected by DELETE were zero,
-                        // and yet the result was not null previously, it indicates this DELETE lost
-                        // a concurrent request race. Hence, empty the non-null result.
                         if (resultProperties is not null
                             && resultProperties.TryGetValue(nameof(DbDataReader.RecordsAffected), out object? value)
-                            && Convert.ToInt32(value) == 0
-                            && result is not null && result.Item1 is not null)
+                            && Convert.ToInt32(value) == 0)
+                        {
+                            // the result was not null previously, it indicates this DELETE lost
+                            // a concurrent request race. Hence, empty the non-null result.
+                            if (result is not null && result.Item1 is not null)
+                            {
+
+                                result = new Tuple<JsonDocument?, IMetadata?>(
+                                    default(JsonDocument),
+                                    PaginationMetadata.MakeEmptyPaginationMetadata());
+                            }
+                            else if (context.Selection.Type.TypeName() == GraphQLUtils.DB_OPERATION_RESULT_TYPE)
+                            {
+                                // no record affected but db call ran successfully.
+                                result = GetDbOperationResultJsonDocument("item not found");
+                            }
+                        }
+                        else if (context.Selection.Type.TypeName() == GraphQLUtils.DB_OPERATION_RESULT_TYPE)
                         {
-                            result = new Tuple<JsonDocument?, IMetadata?>(
-                                default(JsonDocument),
-                                PaginationMetadata.MakeEmptyPaginationMetadata());
+                            result = GetDbOperationResultJsonDocument("success");
                         }
                     }
                     else
@@ -158,17 +168,24 @@ public class SqlMutationEngine : IMutationEngine
 
                         // When read permission is not configured, an error response is returned. So, the mutation result needs to
                         // be computed only when the read permission is configured.
-                        if (isReadPermissionConfigured && mutationResultRow is not null && mutationResultRow.Columns.Count > 0
-                            && !context.Selection.Type.IsScalarType())
+                        if (isReadPermissionConfigured)
                         {
-                            // Because the GraphQL mutation result set columns were exposed (mapped) column names,
-                            // the column names must be converted to backing (source) column names so the
-                            // PrimaryKeyPredicates created in the SqlQueryStructure created by the query engine
-                            // represent database column names.
-                            result = await queryEngine.ExecuteAsync(
-                                        context,
-                                        GetBackingColumnsFromCollection(entityName: entityName, parameters: mutationResultRow.Columns, sqlMetadataProvider: sqlMetadataProvider),
-                                        dataSourceName);
+                            if (mutationResultRow is not null && mutationResultRow.Columns.Count > 0
+                                && !context.Selection.Type.IsScalarType())
+                            {
+                                // Because the GraphQL mutation result set columns were exposed (mapped) column names,
+                                // the column names must be converted to backing (source) column names so the
+                                // PrimaryKeyPredicates created in the SqlQueryStructure created by the query engine
+                                // represent database column names.
+                                result = await queryEngine.ExecuteAsync(
+                                            context,
+                                            GetBackingColumnsFromCollection(entityName: entityName, parameters: mutationResultRow.Columns, sqlMetadataProvider: sqlMetadataProvider),
+                                            dataSourceName);
+                            }
+                            else if (context.Selection.Type.TypeName() == GraphQLUtils.DB_OPERATION_RESULT_TYPE)
+                            {
+                                result = GetDbOperationResultJsonDocument("success");
+                            }
                         }
                     }
 
@@ -829,7 +846,8 @@ private FindRequestContext ConstructFindRequestContext(RestRequestContext contex
 
                 dbResultSetRow = dbResultSet is not null ?
                     (dbResultSet.Rows.FirstOrDefault() ?? new DbResultSetRow()) : null;
-                if (dbResultSetRow is not null && dbResultSetRow.Columns.Count == 0)
+
+                if (dbResultSetRow is not null && dbResultSetRow.Columns.Count == 0 && dbResultSet!.ResultProperties.TryGetValue("RecordsAffected", out object? recordsAffected) && (int)recordsAffected <= 0)
                 {
                     // For GraphQL, insert operation corresponds to Create action.
                     if (operationType is EntityActionOperation.Create)
@@ -1178,5 +1196,18 @@ private string GetValidatedDataSourceName(string dataSourceName)
             // For rest scenarios - no multiple db support. Hence to maintain backward compatibility, we will use the default db.
             return string.IsNullOrEmpty(dataSourceName) ? _runtimeConfigProvider.GetConfig().GetDefaultDataSourceName() : dataSourceName;
         }
+
+        /// <summary>
+        /// Returns DbOperationResult with required result.
+        /// </summary>
+        private static Tuple<JsonDocument?, IMetadata?> GetDbOperationResultJsonDocument(string result)
+        {
+            // Create a JSON object with one field "result" and value result
+            JsonObject jsonObject = new() { { "result", result } };
+
+            return new Tuple<JsonDocument?, IMetadata?>(
+                JsonDocument.Parse(jsonObject.ToString()),
+                PaginationMetadata.MakeEmptyPaginationMetadata());
+        }
     }
 }

src/Core/Services/GraphQLSchemaCreator.cs

@@ -11,6 +11,7 @@
 using Azure.DataApiBuilder.Core.Resolvers.Factories;
 using Azure.DataApiBuilder.Core.Services.MetadataProviders;
 using Azure.DataApiBuilder.Service.Exceptions;
+using Azure.DataApiBuilder.Service.GraphQLBuilder;
 using Azure.DataApiBuilder.Service.GraphQLBuilder.Directives;
 using Azure.DataApiBuilder.Service.GraphQLBuilder.GraphQLTypes;
 using Azure.DataApiBuilder.Service.GraphQLBuilder.Mutations;
@@ -232,6 +233,20 @@ private DocumentNode GenerateSqlGraphQLObjects(RuntimeEntities entities, Diction
                 objectTypes[entityName] = QueryBuilder.AddQueryArgumentsForRelationships(node, inputObjects);
             }
 
+            Dictionary<string, FieldDefinitionNode> fields = new();
+            NameNode nameNode = new(value: GraphQLUtils.DB_OPERATION_RESULT_TYPE);
+            FieldDefinitionNode field = GetDbOperationResultField();
+
+            fields.TryAdd("result", field);
+
+            objectTypes.Add(GraphQLUtils.DB_OPERATION_RESULT_TYPE, new ObjectTypeDefinitionNode(
+                location: null,
+                name: nameNode,
+                description: null,
+                new List<DirectiveNode>(),
+                new List<NamedTypeNode>(),
+                fields.Values.ToImmutableList()));
+
             List<IDefinitionNode> nodes = new(objectTypes.Values);
             return new DocumentNode(nodes);
         }
@@ -267,6 +282,21 @@ private DocumentNode GenerateCosmosGraphQLObjects(HashSet<string> dataSourceName
             return root;
         }
 
+        /// <summary>
+        /// Create and return a default GraphQL result field for a mutation which doesn't
+        /// define a result set and doesn't return any rows.
+        /// </summary>
+        private static FieldDefinitionNode GetDbOperationResultField()
+        {
+            return new(
+                location: null,
+                name: new("result"),
+                description: new StringValueNode("Contains result for mutation execution"),
+                arguments: new List<InputValueDefinitionNode>(),
+                type: new StringType().ToTypeNode(),
+                directives: new List<DirectiveNode>());
+        }
+
         public (DocumentNode, Dictionary<string, InputObjectTypeDefinitionNode>) GenerateGraphQLObjects()
         {
             RuntimeConfig runtimeConfig = _runtimeConfigProvider.GetConfig();

src/Service.GraphQLBuilder/Directives/ModelTypeDirective.cs

@@ -15,7 +15,7 @@ protected override void Configure(IDirectiveTypeDescriptor descriptor)
             descriptor.Name(DirectiveName)
                 .Description("A directive to indicate the type maps to a storable entity not a nested entity.");
 
-            descriptor.Location(DirectiveLocation.Object);
+            descriptor.Location(DirectiveLocation.Object | DirectiveLocation.FieldDefinition);
 
             descriptor.Argument(ModelNameArgument)
                 .Description("Underlying name of the database entity.")