Fix Performance Issue for filtering on varchar columns in MS_SQL database (#2052)

## Why make this change?

- Closes #1973 
- Slow response of DAB when filtering on varchar columns in MS_SQL
compared to other DBs.
- Same query in mssql taking around ~1sec, compared to <50ms in other
DBs for large number of rows, eg. 20 million
- This was happening because the query generated by DAB when runs on Sql
Server for varchar columns, it automatically assumes the filter value of
type nvarchar, due to which it does implicit cast leading to slowness.
- To fix this we needed to identify if the filter is for varchar column,
we explitily specify it to be varchar for faster response.

## What is this change?

- We are modifying the DbCommand and adding extra information that is
sent to SqlClient.
- We introduced a new property called `SqlDbType`, By Default for String
DbType, SqlDbType was considered as `nvarchar`. now we are updating this
property to the correct sql type.
- This allows db to directly execute the command without type casting
due to type mismatch.
- Carved out the code which creates DbCommand in `QueryExecuter`, and
added an override method in `MsSqlQueryExecuter`. here we set the
property SqlDbType for the Parameters.

- Link to Discussion which talks about this highlighting the issue :
#1964 (comment)
- Added a new property SqlDbType in Column class, currently we used to
have only DotNet DbType which doesn't differenciate between varchar and
Nvarchar. So, adding this SqlDbType gave us the freedom to modify query
based on the sql db type.

## How was this tested?

- Since, this is a performance issue and only a new column got
introduced. The main aim here was to make sure excising functionality
doesn't break.
- Added few more extra tests.

The performance difference is more visible when there are large number
of rows. So, I added 20 million records and ran perf test.
Below are the performance results with avg response time:

|  | varchar(before) | nvarchar(before |
|---|---|---|
| GraphQL | ~734ms | ~23ms |
| REST | ~655ms | ~24ms |

|  | varchar(after) | nvarchar(after) |
|---|---|---|
| GraphQL | ~18ms | ~23ms |
| REST | ~18ms | ~25ms |

## BEFORE CHANGE
### GRAPHQL

#### VARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/6b8084be-1863-445a-84d3-4a81faaa5a70)

![image](https://github.com/Azure/data-api-builder/assets/102276754/9d538475-3ab5-4b3e-bd07-ae148a996332)

#### NVARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/30ce863b-a4ff-4348-89ce-80b92d701f6e)

![image](https://github.com/Azure/data-api-builder/assets/102276754/17950827-634c-485f-b7ba-189d71991949)

### REST
#### VARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/8db4c7dc-5a70-4b06-8e11-7a8607076926)

#### NVARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/b5316d57-4735-4c93-8df4-210a08fa4904)

## AFTER CHANGE
### GRAPHQL

#### VARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/95fe1f55-9a7a-4db5-852b-7a3418ce2a57)

![image](https://github.com/Azure/data-api-builder/assets/102276754/85254053-4e5a-432a-b727-89e3ced8686f)

#### NVARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/0e741014-162a-4207-bae5-eeae26529b27)

![image](https://github.com/Azure/data-api-builder/assets/102276754/61945777-9db8-47a3-8ee6-6cc53ad82e3c)

### REST
#### VARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/808c64d2-67c6-4c4b-9559-8c80c5431ca6)

#### NVARCHAR

![image](https://github.com/Azure/data-api-builder/assets/102276754/d0217693-d2e5-4822-af1b-ac56cb668517)

---------

Co-authored-by: Aniruddh Munde <anmunde@microsoft.com>
Co-authored-by: Sean Leonard <sean.leonard@microsoft.com>

Author: 3 people

src/Config/DatabasePrimitives/DatabaseObject.cs

@@ -130,6 +130,7 @@ public class ParameterDefinition
 {
     public Type SystemType { get; set; } = null!;
     public DbType? DbType { get; set; }
+    public SqlDbType? SqlDbType { get; set; }
     public bool HasConfigDefault { get; set; }
     public object? ConfigDefaultValue { get; set; }
 }
@@ -208,6 +209,24 @@ public bool IsAnyColumnNullable(List<string> columnsToCheck)
 
         return null;
     }
+
+    /// <summary>
+    /// Method to get the SqlDbType for:
+    /// 1. column for table/view,
+    /// 2. parameter for stored procedure.
+    /// </summary>
+    /// <param name="paramName">The parameter whose SqlDbType is to be determined.
+    /// For table/view paramName refers to the backingColumnName if aliases are used.</param>
+    /// <returns>SqlDbType for the parameter.</returns>
+    public virtual SqlDbType? GetSqlDbTypeForParam(string paramName)
+    {
+        if (Columns.TryGetValue(paramName, out ColumnDefinition? columnDefinition))
+        {
+            return columnDefinition.SqlDbType;
+        }
+
+        return null;
+    }
 }
 
 /// <summary>
@@ -235,6 +254,7 @@ public class ColumnDefinition
     /// </summary>
     public Type SystemType { get; set; } = typeof(object);
     public DbType? DbType { get; set; }
+    public SqlDbType? SqlDbType { get; set; }
     public bool HasDefault { get; set; }
     public bool IsAutoGenerated { get; set; }
     public bool IsNullable { get; set; }

src/Core/Models/DbConnectionParam.cs

@@ -10,10 +10,11 @@ namespace Azure.DataApiBuilder.Core.Models;
 /// </summary>
 public class DbConnectionParam
 {
-    public DbConnectionParam(object? value, DbType? dbType = null)
+    public DbConnectionParam(object? value, DbType? dbType = null, SqlDbType? sqlDbType = null)
     {
         Value = value;
         DbType = dbType;
+        SqlDbType = sqlDbType;
     }
 
     /// <summary>
@@ -26,4 +27,8 @@ public DbConnectionParam(object? value, DbType? dbType = null)
     // identically and then implicit conversion cannot happen.
     // For more details refer: https://github.com/Azure/data-api-builder/pull/1442.
     public DbType? DbType { get; set; }
+
+    // This is being made nullable
+    // because it's not populated for DB's other than MSSQL.
+    public SqlDbType? SqlDbType { get; set; }
 }

src/Core/Models/GraphQLFilterParsers.cs

@@ -242,12 +242,12 @@ public Predicate Parse(
                             ParseScalarType(
                                 ctx,
                                 argumentSchema: filterArgumentObject.Fields[name],
-                                backingColumnName,
-                                subfields,
-                                schemaName,
-                                sourceName,
-                                sourceAlias,
-                                queryStructure.MakeDbConnectionParam)));
+                                fieldName: backingColumnName,
+                                fields: subfields,
+                                schemaName: schemaName,
+                                tableName: sourceName,
+                                tableAlias: sourceAlias,
+                                processLiterals: queryStructure.MakeDbConnectionParam)));
                 }
             }
         }
@@ -476,7 +476,7 @@ public HttpContext GetHttpContextFromMiddlewareContext(IMiddlewareContext ctx)
     /// </summary>
     /// <param name="ctx">The GraphQL context, used to get the query variables</param>
     /// <param name="argumentSchema">An IInputField object which describes the schema of the scalar input argument (e.g. IntFilterInput)</param>
-    /// <param name="name">The name of the field</param>
+    /// <param name="fieldName">The name of the field</param>
     /// <param name="fields">The subfields of the scalar field</param>
     /// <param name="schemaName">The db schema name to which the table belongs</param>
     /// <param name="tableName">The name of the table underlying the *FilterInput being processed</param>
@@ -485,14 +485,14 @@ public HttpContext GetHttpContextFromMiddlewareContext(IMiddlewareContext ctx)
     private static Predicate ParseScalarType(
         IMiddlewareContext ctx,
         IInputField argumentSchema,
-        string name,
+        string fieldName,
         List<ObjectFieldNode> fields,
         string schemaName,
         string tableName,
         string tableAlias,
         Func<object, string?, string> processLiterals)
     {
-        Column column = new(schemaName, tableName, columnName: name, tableAlias);
+        Column column = new(schemaName, tableName, columnName: fieldName, tableAlias);
 
         return FieldFilterParser.Parse(ctx, argumentSchema, column, fields, processLiterals);
     }

src/Core/Parsers/ODataASTVisitor.cs

@@ -277,6 +277,7 @@ private void PopulateDbTypeForProperty(BinaryOperatorNode nodeIn)
             string? paramName = BaseQueryStructure.GetEncodedParamName(_struct.Counter.Current() - 1);
             _metadataProvider.TryGetBackingColumn(_struct.EntityName, propertyNode.Property.Name, out string? backingColumnName);
             _struct.Parameters[paramName].DbType = _struct.GetUnderlyingSourceDefinition().Columns[backingColumnName!].DbType;
+            _struct.Parameters[paramName].SqlDbType = _struct.GetUnderlyingSourceDefinition().Columns[backingColumnName!].SqlDbType;
         }
 
         /// <summary>

src/Core/Resolvers/BaseQueryStructure.cs

@@ -107,13 +107,16 @@ public BaseQueryStructure(
         ///  Add parameter to Parameters and return the name associated with it
         /// </summary>
         /// <param name="value">Value to be assigned to parameter, which can be null for nullable columns.</param>
-        /// <param name="paramName"> The name of the parameter - column name for table/views or parameter name for stored procedures.</param>
+        /// <param name="paramName"> The name of the parameter - backing column name for table/views or parameter name for stored procedures.</param>
         public virtual string MakeDbConnectionParam(object? value, string? paramName = null)
         {
             string encodedParamName = GetEncodedParamName(Counter.Next());
             if (!string.IsNullOrEmpty(paramName))
             {
-                Parameters.Add(encodedParamName, new(value, GetUnderlyingSourceDefinition().GetDbTypeForParam(paramName)));
+                Parameters.Add(encodedParamName,
+                    new(value,
+                        dbType: GetUnderlyingSourceDefinition().GetDbTypeForParam(paramName),
+                        sqlDbType: GetUnderlyingSourceDefinition().GetSqlDbTypeForParam(paramName)));
             }
             else
             {

src/Core/Resolvers/MsSqlQueryExecutor.cs

@@ -305,12 +305,51 @@ public override async Task<DbResultSet> GetMultipleResultSetsIfAnyAsync(
             return dbResultSet;
         }
 
+        /// <inheritdoc />
+        public override SqlCommand PrepareDbCommand(
+            SqlConnection conn,
+            string sqltext,
+            IDictionary<string, DbConnectionParam> parameters,
+            HttpContext? httpContext,
+            string dataSourceName)
+        {
+            SqlCommand cmd = conn.CreateCommand();
+            cmd.CommandType = CommandType.Text;
+
+            // Add query to send user data from DAB to the underlying database to enable additional security the user might have configured
+            // at the database level.
+            string sessionParamsQuery = GetSessionParamsQuery(httpContext, parameters, dataSourceName);
+
+            cmd.CommandText = sessionParamsQuery + sqltext;
+            if (parameters is not null)
+            {
+                foreach (KeyValuePair<string, DbConnectionParam> parameterEntry in parameters)
+                {
+                    SqlParameter parameter = cmd.CreateParameter();
+                    parameter.ParameterName = parameterEntry.Key;
+                    parameter.Value = parameterEntry.Value.Value ?? DBNull.Value;
+                    PopulateDbTypeForParameter(parameterEntry, parameter);
+                    cmd.Parameters.Add(parameter);
+                }
+            }
+
+            return cmd;
+        }
+
         /// <inheritdoc/>
-        public override void PopulateDbTypeForParameter(KeyValuePair<string, DbConnectionParam> parameterEntry, DbParameter parameter)
+        public static void PopulateDbTypeForParameter(KeyValuePair<string, DbConnectionParam> parameterEntry, SqlParameter parameter)
         {
-            if (parameterEntry.Value is not null && parameterEntry.Value.DbType is not null)
+            if (parameterEntry.Value is not null)
             {
-                parameter.DbType = (DbType)parameterEntry.Value.DbType;
+                if (parameterEntry.Value.DbType is not null)
+                {
+                    parameter.DbType = (DbType)parameterEntry.Value.DbType;
+                }
+
+                if (parameterEntry.Value.SqlDbType is not null)
+                {
+                    parameter.SqlDbType = (SqlDbType)parameterEntry.Value.SqlDbType;
+                }
             }
         }
     }

src/Core/Resolvers/QueryExecutor.cs

@@ -153,25 +153,7 @@ public QueryExecutor(DbExceptionParser dbExceptionParser,
             List<string>? args = null)
         {
             await conn.OpenAsync();
-            DbCommand cmd = conn.CreateCommand();
-            cmd.CommandType = CommandType.Text;
-
-            // Add query to send user data from DAB to the underlying database to enable additional security the user might have configured
-            // at the database level.
-            string sessionParamsQuery = GetSessionParamsQuery(httpContext, parameters, dataSourceName);
-
-            cmd.CommandText = sessionParamsQuery + sqltext;
-            if (parameters is not null)
-            {
-                foreach (KeyValuePair<string, DbConnectionParam> parameterEntry in parameters)
-                {
-                    DbParameter parameter = cmd.CreateParameter();
-                    parameter.ParameterName = parameterEntry.Key;
-                    parameter.Value = parameterEntry.Value.Value ?? DBNull.Value;
-                    PopulateDbTypeForParameter(parameterEntry, parameter);
-                    cmd.Parameters.Add(parameter);
-                }
-            }
+            DbCommand cmd = PrepareDbCommand(conn, sqltext, parameters, httpContext, dataSourceName);
 
             try
             {
@@ -197,6 +179,45 @@ public QueryExecutor(DbExceptionParser dbExceptionParser,
             }
         }
 
+        /// <summary>
+        /// Prepares a database command for execution.
+        /// </summary>
+        /// <param name="conn">Connection object used to connect to database.</param>
+        /// <param name="sqltext">Sql text to be executed.</param>
+        /// <param name="parameters">The parameters used to execute the SQL text.</param>
+        /// <param name="httpContext">Current user httpContext.</param>
+        /// <param name="dataSourceName">The name of the data source.</param>
+        /// <returns>A DbCommand object ready for execution.</returns>
+        public virtual DbCommand PrepareDbCommand(
+            TConnection conn,
+            string sqltext,
+            IDictionary<string, DbConnectionParam> parameters,
+            HttpContext? httpContext,
+            string dataSourceName)
+        {
+            DbCommand cmd = conn.CreateCommand();
+            cmd.CommandType = CommandType.Text;
+
+            // Add query to send user data from DAB to the underlying database to enable additional security the user might have configured
+            // at the database level.
+            string sessionParamsQuery = GetSessionParamsQuery(httpContext, parameters, dataSourceName);
+
+            cmd.CommandText = sessionParamsQuery + sqltext;
+            if (parameters is not null)
+            {
+                foreach (KeyValuePair<string, DbConnectionParam> parameterEntry in parameters)
+                {
+                    DbParameter parameter = cmd.CreateParameter();
+                    parameter.ParameterName = parameterEntry.Key;
+                    parameter.Value = parameterEntry.Value.Value ?? DBNull.Value;
+                    PopulateDbTypeForParameter(parameterEntry, parameter);
+                    cmd.Parameters.Add(parameter);
+                }
+            }
+
+            return cmd;
+        }
+
         /// <inheritdoc />
         public virtual string GetSessionParamsQuery(HttpContext? httpContext, IDictionary<string, DbConnectionParam> parameters, string dataSourceName = "")
         {

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

@@ -124,6 +124,7 @@ public void AddNullifiedUnspecifiedFields(
         /// <summary>
         /// Get column type from table underlying the query structure
         /// </summary>
+        /// <param name="columnName">backing column name</param>
         public Type GetColumnSystemType(string columnName)
         {
             if (GetUnderlyingSourceDefinition().Columns.TryGetValue(columnName, out ColumnDefinition? column))
@@ -275,8 +276,8 @@ protected static IEnumerable<Predicate> CreateJoinPredicates(
                     (leftColumnName, rightColumnName) =>
                     {
                         // no table name or schema here is needed because this is a subquery that joins on table alias
-                        Column leftColumn = new(tableSchema: string.Empty, tableName: string.Empty, leftColumnName, leftTableAlias);
-                        Column rightColumn = new(tableSchema: string.Empty, tableName: string.Empty, rightColumnName, rightTableAlias);
+                        Column leftColumn = new(tableSchema: string.Empty, tableName: string.Empty, columnName: leftColumnName, tableAlias: leftTableAlias);
+                        Column rightColumn = new(tableSchema: string.Empty, tableName: string.Empty, columnName: rightColumnName, tableAlias: rightTableAlias);
                         return new Predicate(
                             new PredicateOperand(leftColumn),
                             PredicateOperation.Equal,

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

@@ -159,7 +159,13 @@ private Predicate CreatePredicateForParam(KeyValuePair<string, object?> param)
             Predicate predicate;
             // since we have already validated param we know backing column exists
             MetadataProvider.TryGetBackingColumn(EntityName, param.Key, out string? backingColumn);
-            if (param.Value is null && !sourceDefinition.Columns[backingColumn!].IsNullable)
+            if (backingColumn is null)
+            {
+                // If param.Key was not present in the ExposedToBackingColumnMap then provided param.Key is already the backing column
+                backingColumn = param.Key;
+            }
+
+            if (param.Value is null && !sourceDefinition.Columns[backingColumn].IsNullable)
             {
                 throw new DataApiBuilderException(
                     message: $"Cannot set argument {param.Key} to null.",
@@ -170,7 +176,7 @@ private Predicate CreatePredicateForParam(KeyValuePair<string, object?> param)
             {
                 predicate = new(
                     new PredicateOperand(
-                        new Column(tableSchema: DatabaseObject.SchemaName, tableName: DatabaseObject.Name, backingColumn!)),
+                        new Column(tableSchema: DatabaseObject.SchemaName, tableName: DatabaseObject.Name, backingColumn)),
                     PredicateOperation.Equal,
                     new PredicateOperand($"{MakeDbConnectionParam(null, backingColumn)}")
                 );

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

@@ -128,6 +128,7 @@ private void PopulateColumns(
                 {
                     // since we have already validated mutationParams we know backing column exists
                     MetadataProvider.TryGetBackingColumn(EntityName, param.Key, out string? backingColumn);
+
                     // Create Parameter and map it to column for downstream logic to utilize.
                     string paramIdentifier;
                     if (param.Value is not null)