diff --git a/src/Core/Models/DbConnectionParam.cs b/src/Core/Models/DbConnectionParam.cs index 0c2c54a5e0..d902022b46 100644 --- a/src/Core/Models/DbConnectionParam.cs +++ b/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; } + + /// + /// Whether the database should infer this parameter's native type from its SQL context. + /// + public bool UseDatabaseTypeInference { get; set; } } diff --git a/src/Core/Resolvers/DWSqlQueryBuilder.cs b/src/Core/Resolvers/DWSqlQueryBuilder.cs index 7157a13b32..f908c21d38 100644 --- a/src/Core/Resolvers/DWSqlQueryBuilder.cs +++ b/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;" + diff --git a/src/Core/Resolvers/MsSqlQueryBuilder.cs b/src/Core/Resolvers/MsSqlQueryBuilder.cs index 118857e701..d689ae0717 100644 --- a/src/Core/Resolvers/MsSqlQueryBuilder.cs +++ b/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;" + diff --git a/src/Core/Resolvers/PostgreSqlExecutor.cs b/src/Core/Resolvers/PostgreSqlExecutor.cs index 4130cd1378..7852be09c2 100644 --- a/src/Core/Resolvers/PostgreSqlExecutor.cs +++ b/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); } + /// + public override void PopulateDbTypeForParameter( + KeyValuePair parameterEntry, + DbParameter parameter) + { + if (parameterEntry.Value.UseDatabaseTypeInference && parameter is NpgsqlParameter npgsqlParameter) + { + npgsqlParameter.NpgsqlDbType = NpgsqlDbType.Unknown; + } + } + /// public override async Task GetMultipleResultSetsIfAnyAsync( DbDataReader dbDataReader, List? 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); diff --git a/src/Core/Resolvers/PostgresQueryBuilder.cs b/src/Core/Resolvers/PostgresQueryBuilder.cs index 9a768f82a0..9ff13f0aee 100644 --- a/src/Core/Resolvers/PostgresQueryBuilder.cs +++ b/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 lockComponents = new() + { + $"'{EscapeSqlLiteral(structure.DatabaseObject.SchemaName)}'", + $"'{EscapeSqlLiteral(structure.DatabaseObject.Name)}'" + }; + List primaryKeys = structure.PrimaryKey(); + + if (primaryKeys.All(primaryKey => IsRepresentationStableKey(structure.GetColumnDefinition(primaryKey)))) + { + Dictionary 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}"; } } + /// + /// 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. + /// + 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); + } + /// /// Build list of LabelledColumns as: /// "{label1}", "{label2}" ... diff --git a/src/Core/Resolvers/Sql Query Structures/SqlUpsertQueryStructure.cs b/src/Core/Resolvers/Sql Query Structures/SqlUpsertQueryStructure.cs index 46a5713ac8..d13a57250a 100644 --- a/src/Core/Resolvers/Sql Query Structures/SqlUpsertQueryStructure.cs +++ b/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 diff --git a/src/Service.Tests/DatabaseSchema-PostgreSql.sql b/src/Service.Tests/DatabaseSchema-PostgreSql.sql index 77edd4a823..854ca48498 100644 --- a/src/Service.Tests/DatabaseSchema-PostgreSql.sql +++ b/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, diff --git a/src/Service.Tests/SqlTests/RestApiTests/MsSqlUpsertConcurrencyTests.cs b/src/Service.Tests/SqlTests/RestApiTests/MsSqlUpsertConcurrencyTests.cs new file mode 100644 index 0000000000..737aab6746 --- /dev/null +++ b/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 +{ + /// + /// Concurrent same-key PUT/PATCH upsert coverage for SQL Server. + /// + [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"; + } + } +} diff --git a/src/Service.Tests/SqlTests/RestApiTests/PostgreSqlUpsertConcurrencyTests.cs b/src/Service.Tests/SqlTests/RestApiTests/PostgreSqlUpsertConcurrencyTests.cs new file mode 100644 index 0000000000..c28bb34b17 --- /dev/null +++ b/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 +{ + /// + /// Concurrent same-key PUT/PATCH upsert coverage for PostgreSQL. + /// + [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 + { + 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}"; + } + + /// + /// Values with different trailing-space representations compare equal for a character(n) key and + /// must therefore be serialized as the same logical key. + /// + [TestMethod] + public async Task ConcurrentUpsertsSerializeDatabaseEqualFixedWidthKeys() + { + for (int iteration = 0; iteration < 8; iteration++) + { + string key = $"K{iteration:D3}"; + string[] databaseEqualKeys = { key, key + " ", key + " ", key + " " }; + Task[] 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 SendFixedWidthKeyUpsertAsync(string key, int value) + { + HttpRequestMessage request = new( + HttpMethod.Put, + $"api/{FIXED_WIDTH_KEY_ENTITY}/id/{Uri.EscapeDataString(key)}") + { + Content = JsonContent.Create(new Dictionary { { "value", value } }) + }; + + request.Headers.Add( + AuthenticationOptions.CLIENT_PRINCIPAL_HEADER, + AuthTestHelper.CreateAppServiceEasyAuthToken( + roleClaimType: AuthenticationOptions.ROLE_CLAIM_TYPE, + additionalClaims: new List + { + new() { Typ = AuthenticationOptions.ROLE_CLAIM_TYPE, Val = "authenticated" } + })); + request.Headers.Add(AuthorizationResolver.CLIENT_ROLE_HEADER, "authenticated"); + + return HttpClient.SendAsync(request); + } + } +} diff --git a/src/Service.Tests/SqlTests/RestApiTests/UpsertConcurrencyTestBase.cs b/src/Service.Tests/SqlTests/RestApiTests/UpsertConcurrencyTestBase.cs new file mode 100644 index 0000000000..5e1782b822 --- /dev/null +++ b/src/Service.Tests/SqlTests/RestApiTests/UpsertConcurrencyTestBase.cs @@ -0,0 +1,142 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +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 +{ + /// + /// Concurrent same-key PUT/PATCH regression coverage shared by relational providers whose REST + /// mutation fixtures support upserts. Each request targets an initially missing composite key. + /// + [TestClass] + public abstract class UpsertConcurrencyTestBase : RestApiTestBase + { + private const int CONCURRENT_REQUEST_COUNT = 4; + private const int ITERATION_COUNT = 8; + + public override string GetQuery(string key) + { + return string.Empty; + } + + protected abstract string GetRowCountQuery(int pieceId); + + [TestCleanup] + public async Task TestCleanup() + { + await ResetDbStateAsync(); + } + + /// + /// Concurrent PUT upserts for one missing key must result in one insert and only updates after it. + /// + [TestMethod] + public async Task ConcurrentPutOnSameMissingPrimaryKeyResolvesCleanly() + { + await RunConcurrentSameKeyUpsertsAsync(HttpMethod.Put, startingPieceId: 9000); + } + + /// + /// Concurrent insert-capable PATCH upserts for one missing key must result in one insert and only + /// updates after it. + /// + [TestMethod] + public async Task ConcurrentPatchOnSameMissingPrimaryKeyResolvesCleanly() + { + await RunConcurrentSameKeyUpsertsAsync(HttpMethod.Patch, startingPieceId: 9100); + } + + private async Task RunConcurrentSameKeyUpsertsAsync(HttpMethod method, int startingPieceId) + { + for (int iteration = 0; iteration < ITERATION_COUNT; iteration++) + { + int pieceId = startingPieceId + iteration; + string primaryKeyRoute = $"categoryid/0/pieceid/{pieceId}"; + Task[] requests = Enumerable.Range(1, CONCURRENT_REQUEST_COUNT) + .Select(requestNumber => SendUpsertAsync(method, primaryKeyRoute, requestNumber)) + .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, + $"{method} iteration {iteration} (pieceid {pieceId}) failed with " + + $"{(int)response.StatusCode} ({response.StatusCode}). Body: {responseBody}"); + } + + int createdCount = responses.Count(response => response.StatusCode == HttpStatusCode.Created); + int okCount = responses.Count(response => response.StatusCode == HttpStatusCode.OK); + + Assert.AreEqual( + 1, + createdCount, + $"{method} iteration {iteration} (pieceid {pieceId}): exactly one request should create the row."); + Assert.AreEqual( + CONCURRENT_REQUEST_COUNT - 1, + okCount, + $"{method} iteration {iteration} (pieceid {pieceId}): all requests after the insert should update the row."); + + string rowCountJson = await GetDatabaseResultAsync(GetRowCountQuery(pieceId)); + using JsonDocument rowCountDocument = JsonDocument.Parse(rowCountJson); + int rowCount = rowCountDocument.RootElement.GetProperty("cnt").GetInt32(); + + Assert.AreEqual( + 1, + rowCount, + $"{method} iteration {iteration} (pieceid {pieceId}): exactly one logical row should remain."); + } + finally + { + foreach (HttpResponseMessage response in responses) + { + response.Dispose(); + } + } + } + } + + private static Task SendUpsertAsync( + HttpMethod method, + string primaryKeyRoute, + int requestNumber) + { + string endpoint = $"api/{_Composite_NonAutoGenPK_EntityPath}/{primaryKeyRoute}"; + HttpRequestMessage request = new(method, endpoint) + { + Content = JsonContent.Create(new Dictionary + { + { "categoryName", "SciFi" }, + { "piecesAvailable", requestNumber }, + { "piecesRequired", requestNumber } + }) + }; + + request.Headers.Add( + AuthenticationOptions.CLIENT_PRINCIPAL_HEADER, + AuthTestHelper.CreateAppServiceEasyAuthToken( + roleClaimType: AuthenticationOptions.ROLE_CLAIM_TYPE, + additionalClaims: new List + { + new() { Typ = AuthenticationOptions.ROLE_CLAIM_TYPE, Val = "authenticated" } + })); + request.Headers.Add(AuthorizationResolver.CLIENT_ROLE_HEADER, "authenticated"); + + return HttpClient.SendAsync(request); + } + } +} diff --git a/src/Service.Tests/UnitTests/PostgreSqlQueryExecutorUnitTests.cs b/src/Service.Tests/UnitTests/PostgreSqlQueryExecutorUnitTests.cs index 6039c46a72..48d6a29d34 100644 --- a/src/Service.Tests/UnitTests/PostgreSqlQueryExecutorUnitTests.cs +++ b/src/Service.Tests/UnitTests/PostgreSqlQueryExecutorUnitTests.cs @@ -3,10 +3,12 @@ using System; using System.Collections.Generic; +using System.Data; using System.Threading.Tasks; using Azure.Core; using Azure.DataApiBuilder.Config.ObjectModel; using Azure.DataApiBuilder.Core.Configurations; +using Azure.DataApiBuilder.Core.Models; using Azure.DataApiBuilder.Core.Resolvers; using Azure.Identity; using Microsoft.AspNetCore.Http; @@ -14,6 +16,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using Moq; using Npgsql; +using NpgsqlTypes; namespace Azure.DataApiBuilder.Service.Tests.UnitTests { @@ -32,6 +35,77 @@ public void TestCleanup() TestHelper.UnsetAllDABEnvironmentVariables(); } + /// + /// Verifies that the advisory-lock result emitted by insert-capable upserts is skipped before + /// the existing count and mutation result sets are interpreted. + /// + [TestMethod] + public async Task InsertCapableUpsertSkipsAdvisoryLockResultSet() + { + RuntimeConfigProvider provider = TestHelper.GetRuntimeConfigProvider(TestHelper.GetRuntimeConfigLoader()); + Mock dbExceptionParser = new(provider); + Mock> queryExecutorLogger = new(); + Mock httpContextAccessor = new(); + PostgreSqlQueryExecutor executor = new( + provider, + dbExceptionParser.Object, + queryExecutorLogger.Object, + httpContextAccessor.Object); + + DataTable lockResult = new(); + lockResult.Columns.Add(PostgresQueryBuilder.UPSERT_LOCK_ACQUIRED, typeof(object)); + lockResult.Rows.Add(DBNull.Value); + + DataTable countResult = new(); + countResult.Columns.Add(PostgresQueryBuilder.COUNT_ROWS_WITH_GIVEN_PK, typeof(long)); + countResult.Columns.Add(PostgresQueryBuilder.IS_FALLBACK_TO_UPDATE, typeof(bool)); + countResult.Rows.Add(0L, false); + + DataTable mutationResult = new(); + mutationResult.Columns.Add("id", typeof(int)); + mutationResult.Rows.Add(42); + + using DataTableReader reader = new(new[] { lockResult, countResult, mutationResult }); + + var result = await executor.GetMultipleResultSetsIfAnyAsync(reader); + + Assert.AreEqual(1, result.Rows.Count); + Assert.AreEqual(42, result.Rows[0].Columns["id"]); + } + + /// + /// PostgreSQL upsert key parameters marked for database type inference must be sent without + /// Npgsql's CLR string-to-text type declaration. + /// + [TestMethod] + public void UpsertKeyParameterUsesDatabaseTypeInference() + { + RuntimeConfigProvider provider = TestHelper.GetRuntimeConfigProvider(TestHelper.GetRuntimeConfigLoader()); + Mock dbExceptionParser = new(provider); + Mock> queryExecutorLogger = new(); + Mock httpContextAccessor = new(); + PostgreSqlQueryExecutor executor = new( + provider, + dbExceptionParser.Object, + queryExecutorLogger.Object, + httpContextAccessor.Object); + + DbConnectionParam connectionParam = new("K000 ") + { + UseDatabaseTypeInference = true + }; + KeyValuePair parameterEntry = new("@param0", connectionParam); + NpgsqlParameter parameter = new() + { + ParameterName = parameterEntry.Key, + Value = connectionParam.Value! + }; + + executor.PopulateDbTypeForParameter(parameterEntry, parameter); + + Assert.AreEqual(NpgsqlDbType.Unknown, parameter.NpgsqlDbType); + } + /// /// Validates managed identity token issued ONLY when connection string does not specify password /// diff --git a/src/Service.Tests/UnitTests/RelationalUpsertConcurrencyQueryBuilderTests.cs b/src/Service.Tests/UnitTests/RelationalUpsertConcurrencyQueryBuilderTests.cs new file mode 100644 index 0000000000..cc2000f6d7 --- /dev/null +++ b/src/Service.Tests/UnitTests/RelationalUpsertConcurrencyQueryBuilderTests.cs @@ -0,0 +1,293 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System; +using System.Collections.Generic; +using System.Data; +using Azure.DataApiBuilder.Auth; +using Azure.DataApiBuilder.Config.DatabasePrimitives; +using Azure.DataApiBuilder.Config.ObjectModel; +using Azure.DataApiBuilder.Core.Authorization; +using Azure.DataApiBuilder.Core.Configurations; +using Azure.DataApiBuilder.Core.Models; +using Azure.DataApiBuilder.Core.Resolvers; +using Azure.DataApiBuilder.Core.Services; +using Azure.DataApiBuilder.Core.Services.MetadataProviders; +using Microsoft.AspNetCore.Http; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Moq; + +namespace Azure.DataApiBuilder.Service.Tests.UnitTests +{ + /// + /// Verifies that insert-capable relational upserts serialize the existence decision before choosing + /// between UPDATE and INSERT. Update-only fallback queries do not need the additional serialization. + /// + [TestClass] + public class RelationalUpsertConcurrencyQueryBuilderTests + { + private const string ENTITY_NAME = "Book"; + private const string SCHEMA_NAME = "dbo"; + private const string TABLE_NAME = "books"; + + private delegate void TryGetColumnCallback(string entity, string field, out string? column); + + /// + /// SQL Server must hold an update/key-range lock while deciding whether a composite-key row exists. + /// + [TestMethod] + public void MsSqlInsertCapableUpsertLocksCompleteCompositeKeyBeforeExistenceCheck() + { + SqlUpsertQueryStructure structure = CreateUpsertStructure(DatabaseType.MSSQL, useCompositePrimaryKey: true); + + string query = new MsSqlQueryBuilder().Build(structure); + + const string lockingExistenceCheck = + "FROM [dbo].[books] WITH (UPDLOCK, HOLDLOCK) WHERE [dbo].[books].[tenant_id] = @param0 AND [dbo].[books].[id] = @param1"; + Assert.IsTrue( + query.Contains(lockingExistenceCheck, StringComparison.Ordinal), + $"Expected the SQL Server existence check to lock the complete composite key. Query: {query}"); + AssertAppearsBefore(query, lockingExistenceCheck, "IF @ROWS_TO_UPDATE = 1"); + } + + /// + /// PostgreSQL can use a key-scoped transaction lock for representation-stable key types. The complete + /// composite key must be included in metadata order so upserts for different keys remain concurrent. + /// + [TestMethod] + public void PostgreSqlRepresentationStableKeyUpsertLocksCompleteKeyBeforeExistenceCheck() + { + SqlUpsertQueryStructure structure = CreateUpsertStructure(DatabaseType.PostgreSQL, useCompositePrimaryKey: true); + + string query = new PostgresQueryBuilder().Build(structure); + + const string advisoryLock = + "SELECT pg_advisory_xact_lock(hashtextextended(jsonb_build_array('dbo', 'books', 'id', @param1, 'tenant_id', @param0)::text, 0))"; + Assert.IsTrue( + query.Contains(advisoryLock, StringComparison.Ordinal), + $"Expected the PostgreSQL advisory lock to identify the complete representation-stable key. Query: {query}"); + AssertAppearsBefore(query, advisoryLock, "SELECT COUNT(*) AS cnt_rows_to_update"); + } + + /// + /// PostgreSQL UUID values are converted to Guid before binding, so equivalent textual UUID forms + /// produce the same key-scoped lock resource. + /// + [TestMethod] + public void PostgreSqlGuidKeyUpsertUsesKeyScopedLock() + { + SqlUpsertQueryStructure structure = CreateUpsertStructure( + DatabaseType.PostgreSQL, + useCompositePrimaryKey: false, + primaryKeySystemType: typeof(Guid)); + + string query = new PostgresQueryBuilder().Build(structure); + + const string advisoryLock = + "SELECT pg_advisory_xact_lock(hashtextextended(jsonb_build_array('dbo', 'books', 'id', @param0)::text, 0))"; + Assert.IsTrue( + query.Contains(advisoryLock, StringComparison.Ordinal), + $"Expected the PostgreSQL advisory lock to identify the UUID key. Query: {query}"); + AssertAppearsBefore(query, advisoryLock, "SELECT COUNT(*) AS cnt_rows_to_update"); + } + + /// + /// PostgreSQL string equality can depend on the backing type and collation, so string keys must use + /// the source-scoped fallback rather than deriving a lock from the request representation. + /// + [TestMethod] + public void PostgreSqlStringKeyUpsertFallsBackToSourceLock() + { + SqlUpsertQueryStructure structure = CreateUpsertStructure( + DatabaseType.PostgreSQL, + useCompositePrimaryKey: false, + primaryKeySystemType: typeof(string)); + + string query = new PostgresQueryBuilder().Build(structure); + + const string advisoryLock = + "SELECT pg_advisory_xact_lock(hashtextextended(jsonb_build_array('dbo', 'books')::text, 0))"; + Assert.IsTrue( + query.Contains(advisoryLock, StringComparison.Ordinal), + $"Expected the PostgreSQL advisory lock to fall back to source scope. Query: {query}"); + Assert.IsFalse( + query[..query.IndexOf(';')].Contains("@param", StringComparison.Ordinal), + $"PostgreSQL source lock identity must not depend on string key representations. Query: {query}"); + Assert.IsTrue( + structure.Parameters["@param0"].UseDatabaseTypeInference, + "PostgreSQL string upsert keys must use the backing column's native comparison semantics."); + Assert.IsFalse( + structure.Parameters["@param1"].UseDatabaseTypeInference, + "Non-key string values must retain normal Npgsql parameter typing."); + AssertAppearsBefore(query, advisoryLock, "SELECT COUNT(*) AS cnt_rows_to_update"); + } + + /// + /// Data Warehouse SQL must hold an exclusive source-table lock before the existence decision because + /// configured logical keys are not necessarily backed by an enforced unique constraint. + /// + [TestMethod] + public void DwSqlInsertCapableUpsertLocksSourceTableBeforeExistenceCheck() + { + SqlUpsertQueryStructure structure = CreateUpsertStructure(DatabaseType.DWSQL, useCompositePrimaryKey: true); + + string query = new DwSqlQueryBuilder().Build(structure); + + const string lockingExistenceCheck = + "FROM [dbo].[books] WITH (TABLOCKX, HOLDLOCK) WHERE [dbo].[books].[tenant_id] = @param0 AND [dbo].[books].[id] = @param1"; + Assert.IsTrue( + query.Contains(lockingExistenceCheck, StringComparison.Ordinal), + $"Expected the Data Warehouse SQL existence check to hold an exclusive source-table lock. Query: {query}"); + AssertAppearsBefore(query, lockingExistenceCheck, "IF @ROWS_TO_UPDATE = 1"); + } + + /// + /// An autogenerated primary key makes the upsert update-only, so no insert race exists and the + /// insert-path serialization primitives must not be emitted. + /// + [TestMethod] + public void UpdateOnlyFallbackUpsertsDoNotAcquireInsertSerializationLocks() + { + SqlUpsertQueryStructure msSqlStructure = CreateUpsertStructure(DatabaseType.MSSQL, useCompositePrimaryKey: false, autoGeneratedPrimaryKey: true); + SqlUpsertQueryStructure postgreSqlStructure = CreateUpsertStructure(DatabaseType.PostgreSQL, useCompositePrimaryKey: false, autoGeneratedPrimaryKey: true); + SqlUpsertQueryStructure dwSqlStructure = CreateUpsertStructure(DatabaseType.DWSQL, useCompositePrimaryKey: false, autoGeneratedPrimaryKey: true); + + string msSqlQuery = new MsSqlQueryBuilder().Build(msSqlStructure); + string postgreSqlQuery = new PostgresQueryBuilder().Build(postgreSqlStructure); + string dwSqlQuery = new DwSqlQueryBuilder().Build(dwSqlStructure); + + Assert.IsFalse(msSqlQuery.Contains("UPDLOCK", StringComparison.Ordinal), $"Update-only SQL Server query should not acquire an insert serialization lock. Query: {msSqlQuery}"); + Assert.IsFalse(postgreSqlQuery.Contains("pg_advisory_xact_lock", StringComparison.Ordinal), $"Update-only PostgreSQL query should not acquire an insert serialization lock. Query: {postgreSqlQuery}"); + Assert.IsFalse(dwSqlQuery.Contains("TABLOCKX", StringComparison.Ordinal), $"Update-only Data Warehouse SQL query should not acquire an insert serialization lock. Query: {dwSqlQuery}"); + } + + private static void AssertAppearsBefore(string query, string first, string second) + { + int firstIndex = query.IndexOf(first, StringComparison.Ordinal); + int secondIndex = query.IndexOf(second, StringComparison.Ordinal); + + Assert.IsTrue(firstIndex >= 0, $"Expected query fragment was not found: {first}. Query: {query}"); + Assert.IsTrue(secondIndex >= 0, $"Expected query fragment was not found: {second}. Query: {query}"); + Assert.IsTrue(firstIndex < secondIndex, $"Expected '{first}' to appear before '{second}'. Query: {query}"); + } + + private static SqlUpsertQueryStructure CreateUpsertStructure( + DatabaseType databaseType, + bool useCompositePrimaryKey, + bool autoGeneratedPrimaryKey = false, + Type? primaryKeySystemType = null) + { + primaryKeySystemType ??= typeof(int); + DbType primaryKeyDbType = DbType.Int32; + object primaryKeyValue = 42; + if (primaryKeySystemType == typeof(string)) + { + primaryKeyDbType = DbType.String; + primaryKeyValue = "book-42"; + } + else if (primaryKeySystemType == typeof(Guid)) + { + primaryKeyDbType = DbType.Guid; + primaryKeyValue = Guid.Parse("00112233-4455-6677-8899-aabbccddeeff"); + } + + SourceDefinition sourceDefinition = new() + { + PrimaryKey = useCompositePrimaryKey ? new() { "id", "tenant_id" } : new() { "id" } + }; + sourceDefinition.Columns.Add("id", new ColumnDefinition + { + SystemType = primaryKeySystemType, + DbType = primaryKeyDbType, + IsAutoGenerated = autoGeneratedPrimaryKey + }); + if (useCompositePrimaryKey) + { + sourceDefinition.Columns.Add("tenant_id", new ColumnDefinition + { + SystemType = typeof(int), + DbType = DbType.Int32 + }); + } + + sourceDefinition.Columns.Add("title", new ColumnDefinition + { + SystemType = typeof(string), + DbType = DbType.String, + IsNullable = true + }); + + DatabaseTable dbTable = new(SCHEMA_NAME, TABLE_NAME) + { + TableDefinition = sourceDefinition, + SourceType = EntitySourceType.Table + }; + + Dictionary columnMapping = new() + { + { "id", "id" }, + { "title", "title" } + }; + if (useCompositePrimaryKey) + { + columnMapping.Add("tenant_id", "tenant_id"); + } + + Mock metadataProvider = new(); + metadataProvider.Setup(x => x.EntityToDatabaseObject) + .Returns(new Dictionary { { ENTITY_NAME, dbTable } }); + metadataProvider.Setup(x => x.GetSourceDefinition(ENTITY_NAME)).Returns(sourceDefinition); + metadataProvider.Setup(x => x.GetDatabaseType()).Returns(databaseType); + + string? outColumn; + metadataProvider.Setup(x => x.TryGetBackingColumn(It.IsAny(), It.IsAny(), out outColumn)) + .Callback(new TryGetColumnCallback((string entity, string field, out string? column) + => columnMapping.TryGetValue(field, out column))) + .Returns((string entity, string field, string? column) => columnMapping.ContainsKey(field)); + + string? outExposed; + metadataProvider.Setup(x => x.TryGetExposedColumnName(It.IsAny(), It.IsAny(), out outExposed)) + .Callback(new TryGetColumnCallback((string entity, string field, out string? column) + => columnMapping.TryGetValue(field, out column))) + .Returns((string entity, string field, string? column) => columnMapping.ContainsKey(field)); + + Mock authorizationResolver = new(); + authorizationResolver + .Setup(x => x.ResolveDBPolicy( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(ResolvedDatabasePolicy.Empty); + + RuntimeConfigProvider runtimeConfigProvider = TestHelper.GetRuntimeConfigProvider(TestHelper.GetRuntimeConfigLoader()); + Mock metadataProviderFactory = new(); + GQLFilterParser gQLFilterParser = new(runtimeConfigProvider, metadataProviderFactory.Object); + + DefaultHttpContext httpContext = new(); + httpContext.Request.Headers[AuthorizationResolver.CLIENT_ROLE_HEADER] = "authenticated"; + + Dictionary mutationParams = useCompositePrimaryKey + ? new() + { + { "tenant_id", 7 }, + { "id", primaryKeyValue }, + { "title", "The Hobbit" } + } + : new() + { + { "id", primaryKeyValue }, + { "title", "The Hobbit" } + }; + + return new SqlUpsertQueryStructure( + entityName: ENTITY_NAME, + sqlMetadataProvider: metadataProvider.Object, + authorizationResolver: authorizationResolver.Object, + gQLFilterParser: gQLFilterParser, + mutationParams: mutationParams, + incrementalUpdate: false, + httpContext: httpContext); + } + } +}