Don't cache merge or insert query (Azure#628)

* dont cache insert/merge query

* add test

* update var name

Author: lucyzhang929

src/SqlAsyncCollector.cs

@@ -48,6 +48,12 @@ public override string ToString()
         }
     }
 
+    public enum QueryType
+    {
+        Insert,
+        Merge
+    }
+
     /// <typeparam name="T">A user-defined POCO that represents a row of the user's table</typeparam>
     internal class SqlAsyncCollector<T> : IAsyncCollector<T>, IDisposable
     {
@@ -224,6 +230,13 @@ private async Task UpsertRowsAsync(IEnumerable<T> rows, SqlAttribute attribute,
                     throw ex;
                 }
 
+                IEnumerable<string> bracketedColumnNamesFromItem = GetColumnNamesFromItem(rows.First())
+                    .Where(prop => !tableInfo.PrimaryKeys.Any(k => k.IsIdentity && string.Equals(k.Name, prop, StringComparison.Ordinal))) // Skip any identity columns, those should never be updated
+                    .Select(prop => prop.AsBracketQuotedString());
+                var table = new SqlObject(fullTableName);
+                string mergeOrInsertQuery = tableInfo.QueryType == QueryType.Insert ? TableInformation.GetInsertQuery(table, bracketedColumnNamesFromItem) :
+                    TableInformation.GetMergeQuery(tableInfo.PrimaryKeys, table, bracketedColumnNamesFromItem);
+
                 TelemetryInstance.TrackEvent(TelemetryEventName.UpsertStart, props);
                 this._logger.LogDebugWithThreadId("BEGIN UpsertRowsTransaction");
                 var transactionSw = Stopwatch.StartNew();
@@ -241,7 +254,7 @@ private async Task UpsertRowsAsync(IEnumerable<T> rows, SqlAttribute attribute,
                     {
                         batchCount++;
                         GenerateDataQueryForMerge(tableInfo, batch, out string newDataQuery, out string rowData);
-                        command.CommandText = $"{newDataQuery} {tableInfo.Query};";
+                        command.CommandText = $"{newDataQuery} {mergeOrInsertQuery};";
                         this._logger.LogDebugWithThreadId($"UpsertRowsTransactionBatch - Query={command.CommandText}");
                         par.Value = rowData;
                         await command.ExecuteNonQueryAsync();
@@ -340,12 +353,12 @@ private static void GenerateDataQueryForMerge(TableInformation table, IEnumerabl
                     else
                     {
                         // SQL Server allows 900 bytes per primary key, so use that as a baseline
-                        var combinedPrimaryKey = new StringBuilder(900 * table.PrimaryKeys.Count());
+                        var combinedPrimaryKey = new StringBuilder(900 * table.PrimaryKeyProperties.Count());
                         // Look up primary key of T. Because we're going in the same order of properties every time,
                         // we can assume that if two rows with the same primary key are in the list, they will collide
-                        foreach (PropertyInfo primaryKey in table.PrimaryKeys)
+                        foreach (PropertyInfo primaryKeyProperty in table.PrimaryKeyProperties)
                         {
-                            object value = primaryKey.GetValue(row);
+                            object value = primaryKeyProperty.GetValue(row);
                             // Identity columns are allowed to be optional, so just skip the key if it doesn't exist
                             if (value == null)
                             {
@@ -379,7 +392,9 @@ private static void GenerateDataQueryForMerge(TableInformation table, IEnumerabl
 
         public class TableInformation
         {
-            public IEnumerable<PropertyInfo> PrimaryKeys { get; }
+            public List<PrimaryKey> PrimaryKeys { get; }
+
+            public IEnumerable<PropertyInfo> PrimaryKeyProperties { get; }
 
             /// <summary>
             /// All of the columns, along with their data types, for SQL to use to turn JSON into a table
@@ -392,10 +407,9 @@ public class TableInformation
             public IEnumerable<string> ColumnDefinitions => this.Columns.Select(c => $"{c.Key} {c.Value}");
 
             /// <summary>
-            /// T-SQL merge or insert statement generated from primary keys
-            /// and column names for a specific table.
+            /// Whether to use an insert query or merge query.
             /// </summary>
-            public string Query { get; }
+            public QueryType QueryType { get; }
 
             /// <summary>
             /// Whether at least one of the primary keys on this table is an identity column
@@ -407,11 +421,12 @@ public class TableInformation
             /// </summary>
             public JsonSerializerSettings JsonSerializerSettings { get; }
 
-            public TableInformation(IEnumerable<PropertyInfo> primaryKeys, IDictionary<string, string> columns, string query, bool hasIdentityColumnPrimaryKeys)
+            public TableInformation(List<PrimaryKey> primaryKeys, IEnumerable<PropertyInfo> primaryKeyProperties, IDictionary<string, string> columns, QueryType queryType, bool hasIdentityColumnPrimaryKeys)
             {
                 this.PrimaryKeys = primaryKeys;
+                this.PrimaryKeyProperties = primaryKeyProperties;
                 this.Columns = columns;
-                this.Query = query;
+                this.QueryType = queryType;
                 this.HasIdentityColumnPrimaryKeys = hasIdentityColumnPrimaryKeys;
 
                 // Convert datetime strings to ISO 8061 format to avoid potential errors on the server when converting into a datetime. This
@@ -625,24 +640,19 @@ public static async Task<TableInformation> RetrieveTableInformationAsync(SqlConn
 
                 // If any identity columns or columns with default values aren't included in the object then we have to generate a basic insert since the merge statement expects all primary key
                 // columns to exist. (the merge statement can handle nullable columns though if those exist)
-                bool usingInsertQuery = (hasIdentityColumnPrimaryKeys || hasDefaultColumnPrimaryKeys) && missingPrimaryKeysFromItem.Any();
-
-                IEnumerable<string> bracketedColumnNamesFromItem = objectColumnNames
-                    .Where(prop => !primaryKeys.Any(k => k.IsIdentity && string.Equals(k.Name, prop, StringComparison.Ordinal))) // Skip any identity columns, those should never be updated
-                    .Select(prop => prop.AsBracketQuotedString());
-                string query = usingInsertQuery ? GetInsertQuery(table, bracketedColumnNamesFromItem) : GetMergeQuery(primaryKeys, table, bracketedColumnNamesFromItem);
+                QueryType queryType = (hasIdentityColumnPrimaryKeys || hasDefaultColumnPrimaryKeys) && missingPrimaryKeysFromItem.Any() ? QueryType.Insert : QueryType.Merge;
 
                 tableInfoSw.Stop();
                 var durations = new Dictionary<TelemetryMeasureName, double>()
                 {
                     { TelemetryMeasureName.GetColumnDefinitionsDurationMs, columnDefinitionsSw.ElapsedMilliseconds },
                     { TelemetryMeasureName.GetPrimaryKeysDurationMs, primaryKeysSw.ElapsedMilliseconds }
                 };
-                sqlConnProps.Add(TelemetryPropertyName.QueryType, usingInsertQuery ? "insert" : "merge");
+                sqlConnProps.Add(TelemetryPropertyName.QueryType, queryType.ToString());
                 sqlConnProps.Add(TelemetryPropertyName.HasIdentityColumn, hasIdentityColumnPrimaryKeys.ToString());
                 TelemetryInstance.TrackDuration(TelemetryEventName.GetTableInfo, tableInfoSw.ElapsedMilliseconds, sqlConnProps, durations);
                 logger.LogDebugWithThreadId($"END RetrieveTableInformationAsync Duration={tableInfoSw.ElapsedMilliseconds}ms DB and Table: {sqlConnection.Database}.{fullName}. Primary keys: [{string.Join(",", primaryKeys.Select(pk => pk.Name))}]. SQL Column and Definitions:  [{string.Join(",", columnDefinitionsFromSQL)}] Object columns: [{string.Join(",", objectColumnNames)}]");
-                return new TableInformation(primaryKeyProperties, columnDefinitionsFromSQL, query, hasIdentityColumnPrimaryKeys);
+                return new TableInformation(primaryKeys, primaryKeyProperties, columnDefinitionsFromSQL, queryType, hasIdentityColumnPrimaryKeys);
             }
         }
 

test/Integration/SqlOutputBindingIntegrationTests.cs

@@ -444,5 +444,34 @@ public void AddProductIncorrectCasing(SupportedLanguages lang)
             Assert.Throws<AggregateException>(() => this.SendOutputGetRequest("addproduct-incorrectcasing").Wait());
             Assert.Equal(0, this.ExecuteScalar("SELECT COUNT(*) FROM Products"));
         }
+
+        /// <summary>
+        /// Tests that subsequent upserts work correctly when the object properties are different from the first upsert.
+        ///
+        [Theory]
+        [SqlInlineData()]
+        public void AddProductWithDifferentPropertiesTest(SupportedLanguages lang)
+        {
+            this.StartFunctionHost(nameof(AddProduct), lang);
+
+            var query1 = new Dictionary<string, object>()
+            {
+                { "ProductId", 0 },
+                { "Name", "test" },
+                { "Cost", 100 }
+            };
+
+            var query2 = new Dictionary<string, object>()
+            {
+                { "ProductId", 0 },
+                { "Name", "test2" }
+            };
+
+            this.SendOutputPostRequest("addproduct", JsonConvert.SerializeObject(query1)).Wait();
+            this.SendOutputPostRequest("addproduct", JsonConvert.SerializeObject(query2)).Wait();
+
+            // Verify result
+            Assert.Equal("test2", this.ExecuteScalar($"select Name from Products where ProductId=0"));
+        }
     }
 }