Fix concurrent same-key PUT and PATCH upserts

src/Core/Models/DbConnectionParam.cs

@@ -35,4 +35,9 @@ public DbConnectionParam(object? value, DbType? dbType = null, SqlDbType? sqlDbT
 
     // Nullable integer parameter representing length. nullable for back compatibility and for where its not needed
     public int? Length { get; set; }
+
+    /// <summary>
+    /// Whether the database should infer this parameter's native type from its SQL context.
+    /// </summary>
+    public bool UseDatabaseTypeInference { get; set; }
 }

src/Core/Resolvers/DWSqlQueryBuilder.cs

@@ -442,7 +442,14 @@ public string Build(SqlUpsertQueryStructure structure)
             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}";
+            // Data Warehouse logical keys are not necessarily enforced by a unique constraint. For an
+            // insert-capable upsert, take and hold an exclusive source-table lock before checking whether
+            // the key exists so concurrent requests cannot both choose INSERT. Update-only fallback queries
+            // cannot insert and therefore do not need this additional serialization.
+            string existenceCheckTable = structure.IsFallbackToUpdate
+                ? tableName
+                : $"{tableName} WITH (TABLOCKX, HOLDLOCK)";
+            string queryToGetCountOfRecordWithPK = $"SELECT COUNT(*) as {COUNT_ROWS_WITH_GIVEN_PK} FROM {existenceCheckTable} WHERE {pkPredicates}";
 
             // Query to get the number of records with a given PK.
             string prefixQuery = $"DECLARE @ROWS_TO_UPDATE int;" +

src/Core/Resolvers/MsSqlQueryBuilder.cs

@@ -285,7 +285,14 @@ public string Build(SqlUpsertQueryStructure structure)
             string updateOperations = Build(structure.UpdateOperations, ", ");
             string columnsToBeReturned =
                 MakeOutputColumns(structure.OutputColumns, isUpdateTriggerEnabled ? string.Empty : OutputQualifier.Inserted.ToString());
-            string queryToGetCountOfRecordWithPK = $"SELECT COUNT(*) as {COUNT_ROWS_WITH_GIVEN_PK} FROM {tableName} WHERE {pkPredicates}";
+            // Insert-capable upserts must serialize the existence decision with competing upserts for
+            // the same key. UPDLOCK avoids lock-conversion deadlocks and HOLDLOCK retains the key-range
+            // lock (including a missing-key range) through the ambient transaction. Update-only fallback
+            // queries do not have an insert race and retain the existing locking behavior.
+            string existenceCheckTable = structure.IsFallbackToUpdate
+                ? tableName
+                : $"{tableName} WITH (UPDLOCK, HOLDLOCK)";
+            string queryToGetCountOfRecordWithPK = $"SELECT COUNT(*) as {COUNT_ROWS_WITH_GIVEN_PK} FROM {existenceCheckTable} WHERE {pkPredicates}";
 
             // Query to get the number of records with a given PK.
             string prefixQuery = $"DECLARE @ROWS_TO_UPDATE int;" +

src/Core/Resolvers/PostgreSqlExecutor.cs

@@ -13,6 +13,7 @@
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.Logging;
 using Npgsql;
+using NpgsqlTypes;
 
 namespace Azure.DataApiBuilder.Core.Resolvers
 {
@@ -148,10 +149,40 @@ private static bool ShouldManagedIdentityAccessBeAttempted(NpgsqlConnectionStrin
             return string.IsNullOrEmpty(builder.Password);
         }
 
+        /// <inheritdoc />
+        public override void PopulateDbTypeForParameter(
+            KeyValuePair<string, DbConnectionParam> parameterEntry,
+            DbParameter parameter)
+        {
+            if (parameterEntry.Value.UseDatabaseTypeInference && parameter is NpgsqlParameter npgsqlParameter)
+            {
+                npgsqlParameter.NpgsqlDbType = NpgsqlDbType.Unknown;
+            }
+        }
+
         /// <inheritdoc/>
         public override async Task<DbResultSet> GetMultipleResultSetsIfAnyAsync(
             DbDataReader dbDataReader, List<string>? args = null)
         {
+            // Insert-capable PostgreSQL upserts acquire a transaction-level advisory lock in a separate
+            // first statement. Consume that result before reading the existence count. Keeping the lock
+            // statement separate ensures the count receives a fresh READ COMMITTED snapshot after any
+            // competing same-key transaction has committed.
+            if (Enumerable.Range(0, dbDataReader.FieldCount).Any(
+                ordinal => string.Equals(
+                    dbDataReader.GetName(ordinal),
+                    PostgresQueryBuilder.UPSERT_LOCK_ACQUIRED,
+                    StringComparison.Ordinal)))
+            {
+                if (!await dbDataReader.NextResultAsync())
+                {
+                    throw new DataApiBuilderException(
+                        message: $"Neither insert nor update could be performed.",
+                        statusCode: HttpStatusCode.InternalServerError,
+                        subStatusCode: DataApiBuilderException.SubStatusCodes.UnexpectedError);
+                }
+            }
+
             // RS1: COUNT of rows matching PK (no policy) — used to distinguish
             // "row doesn't exist" from "row exists but policy blocked".
             DbResultSet resultSetWithCountOfRowsWithGivenPk = await ExtractResultSetFromDbDataReaderAsync(dbDataReader);

src/Core/Resolvers/PostgresQueryBuilder.cs

@@ -1,8 +1,10 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System.Data;
 using System.Data.Common;
 using System.Text;
+using Azure.DataApiBuilder.Config.DatabasePrimitives;
 using Azure.DataApiBuilder.Config.ObjectModel;
 using Azure.DataApiBuilder.Core.Models;
 using Npgsql;
@@ -19,6 +21,7 @@ public class PostgresQueryBuilder : BaseSqlQueryBuilder, IQueryBuilder
         private const string UPDATE_UPSERT = "updated";
         public const string COUNT_ROWS_WITH_GIVEN_PK = "cnt_rows_to_update";
         public const string IS_FALLBACK_TO_UPDATE = "is_fallback_to_update";
+        public const string UPSERT_LOCK_ACQUIRED = "___upsert_lock_acquired___";
 
         private static DbCommandBuilder _builder = new NpgsqlCommandBuilder();
 
@@ -133,6 +136,40 @@ public string Build(SqlUpsertQueryStructure structure)
             string pkPredicates = Build(structure.Predicates);
             string isFallbackToUpdateSqlLiteral = structure.IsFallbackToUpdate ? "TRUE" : "FALSE";
 
+            string lockQuery = string.Empty;
+            if (!structure.IsFallbackToUpdate)
+            {
+                // PostgreSQL row locks cannot protect a key that does not exist. Serialize insert-capable
+                // upserts with a transaction-level advisory lock. Use a key-scoped resource only when every
+                // key value is converted to a representation-stable, non-collatable CLR type. Otherwise use
+                // a source-scoped resource because distinct request representations can compare equal under
+                // the backing key's type or collation (for example character(n) padding or nondeterministic
+                // collations).
+                // Keep acquisition in its own statement so the following READ COMMITTED statement obtains
+                // its snapshot only after a competing lock holder has committed.
+                List<string> lockComponents = new()
+                {
+                    $"'{EscapeSqlLiteral(structure.DatabaseObject.SchemaName)}'",
+                    $"'{EscapeSqlLiteral(structure.DatabaseObject.Name)}'"
+                };
+                List<string> primaryKeys = structure.PrimaryKey();
+
+                if (primaryKeys.All(primaryKey => IsRepresentationStableKey(structure.GetColumnDefinition(primaryKey))))
+                {
+                    Dictionary<string, string> primaryKeyParameters = structure.Predicates.ToDictionary(
+                        predicate => predicate.Left!.AsColumn()!.ColumnName,
+                        predicate => predicate.Right.AsString()!);
+
+                    foreach (string primaryKey in primaryKeys)
+                    {
+                        lockComponents.Add($"'{EscapeSqlLiteral(primaryKey)}'");
+                        lockComponents.Add(primaryKeyParameters[primaryKey]);
+                    }
+                }
+
+                lockQuery = $"SELECT pg_advisory_xact_lock(hashtextextended(jsonb_build_array({string.Join(", ", lockComponents)})::text, 0)) AS {UPSERT_LOCK_ACQUIRED}; ";
+            }
+
             // RS1: COUNT of rows matching PK (no policy) — used to distinguish
             // "row doesn't exist" from "row exists but policy blocked" in the executor.
             string countQuery = $"SELECT COUNT(*) AS {COUNT_ROWS_WITH_GIVEN_PK}, " +
@@ -175,10 +212,32 @@ public string Build(SqlUpsertQueryStructure structure)
                     $"SELECT {BuildListOfLabels(structure.OutputColumns)}, {UPSERT_IDENTIFIER_COLUMN_NAME} FROM update_cte UNION ALL " +
                     $"SELECT {BuildListOfLabels(structure.OutputColumns)}, {UPSERT_IDENTIFIER_COLUMN_NAME} FROM insert_cte;";
 
-                return $"{countQuery}; {cteQuery}";
+                return $"{lockQuery}{countQuery}; {cteQuery}";
             }
         }
 
+        /// <summary>
+        /// Returns whether DAB converts the key to a canonical, non-collatable value before binding it.
+        /// Keep this allowlist conservative; unknown types use the correctness-first source lock.
+        /// </summary>
+        private static bool IsRepresentationStableKey(ColumnDefinition columnDefinition)
+        {
+            if (columnDefinition.IsNullable || columnDefinition.IsArrayType)
+            {
+                return false;
+            }
+
+            return (columnDefinition.SystemType == typeof(short) && columnDefinition.DbType == DbType.Int16) ||
+                (columnDefinition.SystemType == typeof(int) && columnDefinition.DbType == DbType.Int32) ||
+                (columnDefinition.SystemType == typeof(long) && columnDefinition.DbType == DbType.Int64) ||
+                (columnDefinition.SystemType == typeof(Guid) && columnDefinition.DbType == DbType.Guid);
+        }
+
+        private static string EscapeSqlLiteral(string value)
+        {
+            return value.Replace("'", "''", StringComparison.Ordinal);
+        }
+
         /// <summary>
         /// Build list of LabelledColumns as:
         /// "{label1}", "{label2}" ...

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

@@ -154,6 +154,16 @@ private void PopulateColumns(
                     // as Update request uses Where clause to target item by PK.
                     if (primaryKeys.Contains(backingColumn!))
                     {
+                        // A CLR string is normally sent by Npgsql as PostgreSQL text. For upsert key
+                        // predicates, that can change the backing column's equality semantics (for
+                        // example, character(n) trailing-space handling). Let PostgreSQL infer the
+                        // native parameter type from the column comparison instead.
+                        if (MetadataProvider.GetDatabaseType() is DatabaseType.PostgreSQL &&
+                            Parameters[paramIdentifier].Value is string)
+                        {
+                            Parameters[paramIdentifier].UseDatabaseTypeInference = true;
+                        }
+
                         PopulateColumnsAndParams(backingColumn!);
 
                         // PK added as predicate for Update Operation

src/Service.Tests/DatabaseSchema-PostgreSql.sql

@@ -20,6 +20,7 @@ DROP TABLE IF EXISTS foo.magazines;
 DROP TABLE IF EXISTS bar.magazines;
 DROP TABLE IF EXISTS stocks_price;
 DROP TABLE IF EXISTS stocks;
+DROP TABLE IF EXISTS fixed_width_key_upsert;
 DROP TABLE IF EXISTS comics;
 DROP TABLE IF EXISTS brokers;
 DROP TABLE IF EXISTS array_type_table;
@@ -137,6 +138,11 @@ CREATE TABLE stocks(
     PRIMARY KEY(categoryid, pieceid)
 );
 
+CREATE TABLE fixed_width_key_upsert(
+    id character(8) PRIMARY KEY,
+    value int NOT NULL
+);
+
 CREATE TABLE stocks_price(
     categoryid int NOT NULL,
     pieceid int NOT NULL,

src/Service.Tests/SqlTests/RestApiTests/MsSqlUpsertConcurrencyTests.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Azure.DataApiBuilder.Service.Tests.SqlTests.RestApiTests
+{
+    /// <summary>
+    /// Concurrent same-key PUT/PATCH upsert coverage for SQL Server.
+    /// </summary>
+    [TestClass, TestCategory(TestCategory.MSSQL)]
+    public class MsSqlUpsertConcurrencyTests : UpsertConcurrencyTestBase
+    {
+        [ClassInitialize]
+        public static async Task SetupAsync(TestContext context)
+        {
+            DatabaseEngine = TestCategory.MSSQL;
+            await InitializeTestFixture();
+        }
+
+        protected override string GetRowCountQuery(int pieceId)
+        {
+            return $"SELECT COUNT(*) AS [cnt] FROM {_Composite_NonAutoGenPK_TableName} " +
+                $"WHERE [categoryid] = 0 AND [pieceid] = {pieceId} " +
+                "FOR JSON PATH, WITHOUT_ARRAY_WRAPPER";
+        }
+    }
+}

src/Service.Tests/SqlTests/RestApiTests/PostgreSqlUpsertConcurrencyTests.cs

@@ -0,0 +1,112 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Json;
+using System.Text.Json;
+using System.Threading.Tasks;
+using Azure.DataApiBuilder.Config.ObjectModel;
+using Azure.DataApiBuilder.Core.Authorization;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using static Azure.DataApiBuilder.Core.AuthenticationHelpers.AppServiceAuthentication;
+
+namespace Azure.DataApiBuilder.Service.Tests.SqlTests.RestApiTests
+{
+    /// <summary>
+    /// Concurrent same-key PUT/PATCH upsert coverage for PostgreSQL.
+    /// </summary>
+    [TestClass, TestCategory(TestCategory.POSTGRESQL)]
+    public class PostgreSqlUpsertConcurrencyTests : UpsertConcurrencyTestBase
+    {
+        private const string FIXED_WIDTH_KEY_ENTITY = "FixedWidthKeyUpsert";
+
+        [ClassInitialize]
+        public static async Task SetupAsync(TestContext context)
+        {
+            DatabaseEngine = TestCategory.POSTGRESQL;
+            await InitializeTestFixture(
+                customEntities: new List<string[]>
+                {
+                    new[] { FIXED_WIDTH_KEY_ENTITY, "fixed_width_key_upsert" }
+                });
+        }
+
+        protected override string GetRowCountQuery(int pieceId)
+        {
+            return "SELECT json_build_object('cnt', COUNT(*)) AS data " +
+                $"FROM {_Composite_NonAutoGenPK_TableName} " +
+                $"WHERE categoryid = 0 AND pieceid = {pieceId}";
+        }
+
+        /// <summary>
+        /// Values with different trailing-space representations compare equal for a character(n) key and
+        /// must therefore be serialized as the same logical key.
+        /// </summary>
+        [TestMethod]
+        public async Task ConcurrentUpsertsSerializeDatabaseEqualFixedWidthKeys()
+        {
+            for (int iteration = 0; iteration < 8; iteration++)
+            {
+                string key = $"K{iteration:D3}";
+                string[] databaseEqualKeys = { key, key + " ", key + "  ", key + "   " };
+                Task<HttpResponseMessage>[] requests = databaseEqualKeys
+                    .Select((databaseEqualKey, index) => SendFixedWidthKeyUpsertAsync(databaseEqualKey, index + 1))
+                    .ToArray();
+
+                HttpResponseMessage[] responses = await Task.WhenAll(requests);
+                try
+                {
+                    foreach (HttpResponseMessage response in responses)
+                    {
+                        string responseBody = await response.Content.ReadAsStringAsync();
+                        Assert.IsTrue(
+                            response.StatusCode is HttpStatusCode.OK or HttpStatusCode.Created,
+                            $"Fixed-width key upsert failed with {(int)response.StatusCode} " +
+                            $"({response.StatusCode}). Body: {responseBody}");
+                    }
+
+                    Assert.AreEqual(1, responses.Count(response => response.StatusCode == HttpStatusCode.Created));
+                    Assert.AreEqual(databaseEqualKeys.Length - 1, responses.Count(response => response.StatusCode == HttpStatusCode.OK));
+
+                    string rowCountJson = await GetDatabaseResultAsync(
+                        $"SELECT json_build_object('cnt', COUNT(*)) AS data FROM fixed_width_key_upsert WHERE id = '{key}'");
+                    using JsonDocument rowCountDocument = JsonDocument.Parse(rowCountJson);
+                    Assert.AreEqual(1, rowCountDocument.RootElement.GetProperty("cnt").GetInt32());
+                }
+                finally
+                {
+                    foreach (HttpResponseMessage response in responses)
+                    {
+                        response.Dispose();
+                    }
+                }
+            }
+        }
+
+        private static Task<HttpResponseMessage> SendFixedWidthKeyUpsertAsync(string key, int value)
+        {
+            HttpRequestMessage request = new(
+                HttpMethod.Put,
+                $"api/{FIXED_WIDTH_KEY_ENTITY}/id/{Uri.EscapeDataString(key)}")
+            {
+                Content = JsonContent.Create(new Dictionary<string, object> { { "value", value } })
+            };
+
+            request.Headers.Add(
+                AuthenticationOptions.CLIENT_PRINCIPAL_HEADER,
+                AuthTestHelper.CreateAppServiceEasyAuthToken(
+                    roleClaimType: AuthenticationOptions.ROLE_CLAIM_TYPE,
+                    additionalClaims: new List<AppServiceClaim>
+                    {
+                        new() { Typ = AuthenticationOptions.ROLE_CLAIM_TYPE, Val = "authenticated" }
+                    }));
+            request.Headers.Add(AuthorizationResolver.CLIENT_ROLE_HEADER, "authenticated");
+
+            return HttpClient.SendAsync(request);
+        }
+    }
+}