From 3a025fafda7fbefc084d4fe7eda938f76af5c12d Mon Sep 17 00:00:00 2001 From: Shay Rojansky Date: Sun, 18 Sep 2022 11:09:20 +0200 Subject: [PATCH 1/3] Implement batching error barrier control Closes #4205 --- src/Npgsql/Internal/NpgsqlConnector.cs | 3 +- src/Npgsql/NpgsqlBatch.cs | 26 +++ src/Npgsql/NpgsqlBatchCommand.cs | 25 +++ src/Npgsql/NpgsqlCommand.cs | 11 +- src/Npgsql/NpgsqlDataReader.cs | 142 ++++++++++--- src/Npgsql/PostgresMinimalDatabaseInfo.cs | 6 +- src/Npgsql/PublicAPI.Unshipped.txt | 4 + src/Npgsql/Util/PGUtil.cs | 12 +- test/Npgsql.Tests/BatchTests.cs | 235 +++++++++++++++++++++- 9 files changed, 421 insertions(+), 43 deletions(-) diff --git a/src/Npgsql/Internal/NpgsqlConnector.cs b/src/Npgsql/Internal/NpgsqlConnector.cs index 4fa274aa46..ea225db10d 100644 --- a/src/Npgsql/Internal/NpgsqlConnector.cs +++ b/src/Npgsql/Internal/NpgsqlConnector.cs @@ -1358,7 +1358,8 @@ internal ValueTask ReadMessage(bool async, DataRowLoadingMode d // an RFQ. Instead, the server closes the connection immediately throw error; } - else if (PostgresErrorCodes.IsCriticalFailure(error, clusterError: false)) + + if (PostgresErrorCodes.IsCriticalFailure(error, clusterError: false)) { // Consider the connection dead throw connector.Break(error); diff --git a/src/Npgsql/NpgsqlBatch.cs b/src/Npgsql/NpgsqlBatch.cs index c60d57654e..0b86bb3164 100644 --- a/src/Npgsql/NpgsqlBatch.cs +++ b/src/Npgsql/NpgsqlBatch.cs @@ -54,6 +54,32 @@ protected override DbTransaction? DbTransaction set => Transaction = (NpgsqlTransaction?)value; } + /// + /// Controls whether to place error barriers between all batch commands within this batch. Default to . + /// + /// + /// + /// By default, any exception in a command causes later commands in the batch to be skipped, and earlier commands to be rolled back. + /// Enabling error barriers ensures that errors do not affect other commands in the batch. + /// + /// + /// Note that if the batch is executed within an explicit transaction, the first error places the transaction in a failed state, + /// causing all later commands to fail in any case. As a result, this option is useful mainly when there is no explicit transaction. + /// + /// + /// At the PostgreSQL wire protocol level, this corresponds to inserting a Sync message between each command, rather than grouping + /// all the batch's commands behind a single terminating Sync. + /// + /// + /// To control error barriers on a command-by-command basis, see . + /// + /// + public bool EnableErrorBarriers + { + get => Command.EnableErrorBarriers; + set => Command.EnableErrorBarriers = value; + } + /// /// Marks all of the batch's result columns as either known or unknown. /// Unknown results column are requested them from PostgreSQL in text format, and Npgsql makes no diff --git a/src/Npgsql/NpgsqlBatchCommand.cs b/src/Npgsql/NpgsqlBatchCommand.cs index 6ec2db4e89..78aedc1f7e 100644 --- a/src/Npgsql/NpgsqlBatchCommand.cs +++ b/src/Npgsql/NpgsqlBatchCommand.cs @@ -32,6 +32,31 @@ public override string CommandText /// public new NpgsqlParameterCollection Parameters { get; } = new(); + /// + /// Appends an error barrier after this batch command. Defaults to the value of on the + /// batch. + /// + /// + /// + /// By default, any exception in a command causes later commands in the batch to be skipped, and earlier commands to be rolled back. + /// Appending an error barrier ensures that errors from this command (or previous ones) won't cause later commands to be skipped, + /// and that errors from later commands won't cause this command (or previous ones) to be rolled back). + /// + /// + /// Note that if the batch is executed within an explicit transaction, the first error places the transaction in a failed state, + /// causing all later commands to fail in any case. As a result, this option is useful mainly when there is no explicit transaction. + /// + /// + /// At the PostgreSQL wire protocol level, this corresponds to inserting a Sync message after this command, rather than grouping + /// all the batch's commands behind a single terminating Sync. + /// + /// + /// Controlling error barriers on a command-by-command basis is an advanced feature, consider enabling error barriers for the entire + /// batch via . + /// + /// + public bool? AppendErrorBarrier { get; set; } + /// /// The number of rows affected or retrieved. /// diff --git a/src/Npgsql/NpgsqlCommand.cs b/src/Npgsql/NpgsqlCommand.cs index 48fc1edd6b..6cd50fe98d 100644 --- a/src/Npgsql/NpgsqlCommand.cs +++ b/src/Npgsql/NpgsqlCommand.cs @@ -76,6 +76,8 @@ public class NpgsqlCommand : DbCommand, ICloneable, IComponent internal static readonly bool EnableSqlRewriting; #endif + internal bool EnableErrorBarriers { get; set; } + static readonly List EmptyParameters = new(); static readonly SingleThreadSynchronizationContext SingleThreadSynchronizationContext = new("NpgsqlRemainingAsyncSendWorker"); @@ -1000,11 +1002,18 @@ await connector.WriteBind( await connector.WriteExecute(0, async, cancellationToken); + if (batchCommand.AppendErrorBarrier ?? EnableErrorBarriers) + await connector.WriteSync(async, cancellationToken); + if (pStatement != null) pStatement.LastUsed = DateTime.UtcNow; } - await connector.WriteSync(async, cancellationToken); + if (InternalBatchCommands.Count == 0 || + !EnableErrorBarriers && InternalBatchCommands[InternalBatchCommands.Count - 1].AppendErrorBarrier != true) + { + await connector.WriteSync(async, cancellationToken); + } if (flush) await connector.Flush(async, cancellationToken); diff --git a/src/Npgsql/NpgsqlDataReader.cs b/src/Npgsql/NpgsqlDataReader.cs index b9ffdd2ba3..83cc3dd504 100644 --- a/src/Npgsql/NpgsqlDataReader.cs +++ b/src/Npgsql/NpgsqlDataReader.cs @@ -9,6 +9,7 @@ using System.IO; using System.Linq; using System.Runtime.CompilerServices; +using System.Runtime.ExceptionServices; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -283,9 +284,24 @@ async Task Read(bool async, CancellationToken cancellationToken = default) throw new ArgumentOutOfRangeException(); } - var msg2 = await ReadMessage(async); - ProcessMessage(msg2); - return msg2.Code == BackendMessageCode.DataRow; + var msg = await ReadMessage(async); + + switch (msg.Code) + { + case BackendMessageCode.DataRow: + ProcessMessage(msg); + return true; + + case BackendMessageCode.CommandComplete: + case BackendMessageCode.EmptyQueryResponse: + ProcessMessage(msg); + if (_statements[StatementIndex].AppendErrorBarrier ?? Command.EnableErrorBarriers) + Expect(await Connector.ReadMessage(async), Connector); + return false; + + default: + throw Connector.UnexpectedMessageReceived(msg.Code); + } } catch { @@ -335,10 +351,11 @@ public override bool NextResult() => (_isSchemaOnly ? NextResultSchemaOnly(false /// A task representing the asynchronous operation. public override Task NextResultAsync(CancellationToken cancellationToken) { - using (NoSynchronizationContextScope.Enter()) - return _isSchemaOnly - ? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken) - : NextResult(async: true, cancellationToken: cancellationToken); + using var _ = NoSynchronizationContextScope.Enter(); + + return _isSchemaOnly + ? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken) + : NextResult(async: true, cancellationToken: cancellationToken); } /// @@ -370,7 +387,12 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo case BackendMessageCode.CommandComplete: case BackendMessageCode.EmptyQueryResponse: ProcessMessage(completedMsg); + + if (_statements[StatementIndex].AppendErrorBarrier ?? Command.EnableErrorBarriers) + Expect(await Connector.ReadMessage(async), Connector); + break; + default: continue; } @@ -472,6 +494,10 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo } ProcessMessage(msg); + + if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) + Expect(await Connector.ReadMessage(async), Connector); + continue; } @@ -494,30 +520,34 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo switch (msg.Code) { case BackendMessageCode.DataRow: + return true; case BackendMessageCode.CommandComplete: - break; + if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) + Expect(await Connector.ReadMessage(async), Connector); + return true; default: throw Connector.UnexpectedMessageReceived(msg.Code); } - - return true; } // There are no more queries, we're done. Read the RFQ. - ProcessMessage(Expect(await Connector.ReadMessage(async), Connector)); + if (_statements.Count == 0 || + !Command.EnableErrorBarriers && _statements[_statements.Count - 1].AppendErrorBarrier != true) + { + Expect(await Connector.ReadMessage(async), Connector); + } + State = ReaderState.Consumed; RowDescription = null; return false; } catch (Exception e) { - State = ReaderState.Consumed; - // Reference the triggering statement from the exception if (e is PostgresException postgresException && StatementIndex >= 0 && StatementIndex < _statements.Count) { postgresException.BatchCommand = _statements[StatementIndex]; - // Prevent the command or batch from by recycled (by the connection) when it's disposed. This is important since + // Prevent the command or batch from being recycled (by the connection) when it's disposed. This is important since // the exception is very likely to escape the using statement of the command, and by that time some other user may // already be using the recycled instance. if (!Command.IsWrappedByBatch) @@ -526,9 +556,8 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo } } - // An error means all subsequent statements were skipped by PostgreSQL. - // If any of them were being prepared, we need to update our bookkeeping to put - // them back in unprepared state. + // For the statement that errored, if it was being prepared we need to update our bookkeeping to put them back in unprepared + // state. for (; StatementIndex < _statements.Count; StatementIndex++) { var statement = _statements[StatementIndex]; @@ -537,6 +566,34 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo statement.IsPreparing = false; statement.PreparedStatement!.AbortPrepare(); } + + if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) + break; + } + + // In normal, non-isolated batching, we've consumed the result set and are done. + // However, if an isolated command was present after the error, we now have to consume the rest of the result set. + // Note that Consume calls NextResult (this method) recursively, the isConsuming flag tells us we're in this mode. + if (StatementIndex == _statements.Count) + { + State = ReaderState.Consumed; + } + else if (!isConsuming) + { + switch (State) + { + case ReaderState.Consumed: + case ReaderState.Closed: + case ReaderState.Disposed: + // The exception may have caused the connector to break (e.g. I/O), and so the reader is already closed. + break; + default: + // We provide Consume with the first exception which we've just caught. + // If it encounters other exceptions while consuming the rest of the result set, it will raise an AggregateException, + // otherwise it will rethrow this first exception. + await Consume(async, firstException: e); + break; + } } throw; @@ -672,8 +729,9 @@ async Task NextResultSchemaOnly(bool async, bool isConsuming = false, Canc // There are no more queries, we're done. Read to the RFQ. if (!_statements.All(s => s.IsPrepared)) { - ProcessMessage(Expect(await Connector.ReadMessage(async), Connector)); + Expect(await Connector.ReadMessage(async), Connector); RowDescription = null; + State = ReaderState.Consumed; } return false; @@ -748,10 +806,6 @@ internal void ProcessMessage(IBackendMessage msg) State = ReaderState.BetweenResults; return; - case BackendMessageCode.ReadyForQuery: - State = ReaderState.Consumed; - return; - default: throw new Exception("Received unexpected backend message of type " + msg.Code); } @@ -901,14 +955,44 @@ public override int FieldCount /// Consumes all result sets for this reader, leaving the connector ready for sending and processing further /// queries /// - async Task Consume(bool async) + async Task Consume(bool async, Exception? firstException = null) { - // Skip over the other result sets. Note that this does tally records affected - // from CommandComplete messages, and properly sets state for auto-prepared statements - if (_isSchemaOnly) - while (await NextResultSchemaOnly(async, isConsuming: true)) {} - else - while (await NextResult(async, isConsuming: true)) {} + var exceptions = firstException is null ? null : new List { firstException }; + + // Skip over the other result sets. Note that this does tally records affected from CommandComplete messages, and properly sets + // state for auto-prepared statements + while (true) + { + try + { + if (!(_isSchemaOnly + ? await NextResultSchemaOnly(async, isConsuming: true) + : await NextResult(async, isConsuming: true))) + { + break; + } + } + catch (Exception e) + { + exceptions ??= new(); + exceptions.Add(e); + } + } + + Debug.Assert(exceptions?.Count != 0); + + switch (exceptions?.Count) + { + case null: + return; + case 1: + ExceptionDispatchInfo.Capture(exceptions[0]).Throw(); + return; + default: + throw new NpgsqlException( + "Multiple exceptions occurred when consuming the result set", + new AggregateException(exceptions)); + } } /// diff --git a/src/Npgsql/PostgresMinimalDatabaseInfo.cs b/src/Npgsql/PostgresMinimalDatabaseInfo.cs index e198c88243..924bb56c2f 100644 --- a/src/Npgsql/PostgresMinimalDatabaseInfo.cs +++ b/src/Npgsql/PostgresMinimalDatabaseInfo.cs @@ -20,7 +20,7 @@ class PostgresMinimalDatabaseInfoFactory : INpgsqlDatabaseInfoFactory class PostgresMinimalDatabaseInfo : PostgresDatabaseInfo { - static PostgresType[]? TypesWithMultiranges, TypesWithoutMultiranges; + static PostgresType[]? _typesWithMultiranges, _typesWithoutMultiranges; static PostgresType[] CreateTypes(bool withMultiranges) => typeof(NpgsqlDbType).GetFields() @@ -50,8 +50,8 @@ static PostgresType[] CreateTypes(bool withMultiranges) protected override IEnumerable GetTypes() => SupportsMultirangeTypes - ? TypesWithMultiranges ??= CreateTypes(withMultiranges: true) - : TypesWithoutMultiranges ??= CreateTypes(withMultiranges: false); + ? _typesWithMultiranges ??= CreateTypes(withMultiranges: true) + : _typesWithoutMultiranges ??= CreateTypes(withMultiranges: false); internal PostgresMinimalDatabaseInfo(NpgsqlConnector conn) : base(conn) diff --git a/src/Npgsql/PublicAPI.Unshipped.txt b/src/Npgsql/PublicAPI.Unshipped.txt index 5e26f00162..321517f286 100644 --- a/src/Npgsql/PublicAPI.Unshipped.txt +++ b/src/Npgsql/PublicAPI.Unshipped.txt @@ -1,4 +1,8 @@ #nullable enable +Npgsql.NpgsqlBatch.EnableErrorBarriers.get -> bool +Npgsql.NpgsqlBatch.EnableErrorBarriers.set -> void +Npgsql.NpgsqlBatchCommand.AppendErrorBarrier.get -> bool? +Npgsql.NpgsqlBatchCommand.AppendErrorBarrier.set -> void Npgsql.NpgsqlLoggingConfiguration static Npgsql.NpgsqlLoggingConfiguration.InitializeLogging(Microsoft.Extensions.Logging.ILoggerFactory! loggerFactory, bool parameterLoggingEnabled = false) -> void *REMOVED*Npgsql.NpgsqlConnection.Settings.get -> Npgsql.NpgsqlConnectionStringBuilder! diff --git a/src/Npgsql/Util/PGUtil.cs b/src/Npgsql/Util/PGUtil.cs index 40e797a2a0..97430badcd 100644 --- a/src/Npgsql/Util/PGUtil.cs +++ b/src/Npgsql/Util/PGUtil.cs @@ -27,14 +27,10 @@ static Statics() [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static T Expect(IBackendMessage msg, NpgsqlConnector connector) - { - if (msg is T asT) - return asT; - - throw connector.Break( - new NpgsqlException($"Received backend message {msg.Code} while expecting {typeof(T).Name}. " + - "Please file a bug.")); - } + => msg is T t + ? t + : throw connector.Break( + new NpgsqlException($"Received backend message {msg.Code} while expecting {typeof(T).Name}. Please file a bug.")); internal static DeferDisposable Defer(Action action) => new(action); internal static DeferDisposable Defer(Action action, T arg) => new(action, arg); diff --git a/test/Npgsql.Tests/BatchTests.cs b/test/Npgsql.Tests/BatchTests.cs index a32b162dad..816e81b775 100644 --- a/test/Npgsql.Tests/BatchTests.cs +++ b/test/Npgsql.Tests/BatchTests.cs @@ -1,4 +1,3 @@ -using Npgsql.Tests.Support; using Npgsql.Util; using NUnit.Framework; using System; @@ -330,6 +329,240 @@ public async Task CloseConnection() #endregion Command behaviors + #region Error barriers + + [Test] + public async Task Batch_with_error_at_start([Values] bool withErrorBarriers) + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (8)") + }, + EnableErrorBarriers = withErrorBarriers + }; + + var exception = Assert.ThrowsAsync(async () => await batch.ExecuteReaderAsync(Behavior))!; + Assert.That(exception.BatchCommand, Is.SameAs(batch.BatchCommands[0])); + + Assert.That(await conn.ExecuteScalarAsync($"SELECT count(*) FROM {table}"), withErrorBarriers + ? Is.EqualTo(1) + : Is.EqualTo(0)); + } + + [Test] + public async Task Batch_with_error_at_end([Values] bool withErrorBarriers) + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new($"INSERT INTO {table} (id) VALUES (8)"), + new("INVALID SQL") + }, + EnableErrorBarriers = withErrorBarriers + }; + + var exception = Assert.ThrowsAsync(async () => await batch.ExecuteReaderAsync(Behavior))!; + Assert.That(exception.BatchCommand, Is.SameAs(batch.BatchCommands[1])); + + Assert.That(await conn.ExecuteScalarAsync($"SELECT count(*) FROM {table}"), withErrorBarriers + ? Is.EqualTo(1) + : Is.EqualTo(0)); + } + + [Test] + public async Task Batch_with_multiple_errors([Values] bool withErrorBarriers) + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new($"INSERT INTO {table} (id) VALUES (8)"), + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (9)"), + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (10)") + }, + EnableErrorBarriers = withErrorBarriers + }; + + if (withErrorBarriers) + { + // A Sync is inserted after each command, so all commands are executed and all exceptions are thrown as an AggregateException + var exception = Assert.ThrowsAsync(async () => await batch.ExecuteReaderAsync(Behavior))!; + var aggregateException = (AggregateException)exception.InnerException!; + Assert.That(((PostgresException)aggregateException.InnerExceptions[0]).BatchCommand, Is.SameAs(batch.BatchCommands[1])); + Assert.That(((PostgresException)aggregateException.InnerExceptions[1]).BatchCommand, Is.SameAs(batch.BatchCommands[3])); + + Assert.That(await conn.ExecuteScalarAsync($"SELECT count(*) FROM {table}"), Is.EqualTo(3)); + } + else + { + // PG skips all commands after the first error; an exception is only raised for the first one, and the entire batch is + // rolled back (implicit transaction). + var exception = Assert.ThrowsAsync(async () => await batch.ExecuteReaderAsync(Behavior))!; + Assert.That(exception.BatchCommand, Is.SameAs(batch.BatchCommands[1])); + + Assert.That(await conn.ExecuteScalarAsync($"SELECT count(*) FROM {table}"), Is.EqualTo(0)); + } + } + + [Test] + public async Task Batch_close_reader_with_multiple_errors([Values] bool withErrorBarriers) + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new("SELECT NULL WHERE 1=0"), + new($"INSERT INTO {table} (id) VALUES (8)"), + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (9)"), + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (10)") + }, + EnableErrorBarriers = withErrorBarriers + }; + + await using var reader = await batch.ExecuteReaderAsync(Behavior); + + if (withErrorBarriers) + { + // A Sync is inserted after each command, so all commands are executed and all exceptions are thrown as an AggregateException + var exception = Assert.ThrowsAsync(async () => await reader.NextResultAsync())!; + var aggregateException = (AggregateException)exception.InnerException!; + Assert.That(((PostgresException)aggregateException.InnerExceptions[0]).BatchCommand, Is.SameAs(batch.BatchCommands[2])); + Assert.That(((PostgresException)aggregateException.InnerExceptions[1]).BatchCommand, Is.SameAs(batch.BatchCommands[4])); + } + else + { + // PG skips all commands after the first error; an exception is only raised for the first one, and the entire batch is + // rolled back (implicit transaction). + var exception = Assert.ThrowsAsync(async () => await reader.NextResultAsync())!; + Assert.That(exception.BatchCommand, Is.SameAs(batch.BatchCommands[2])); + } + } + + [Test] + public async Task Batch_with_result_sets_and_error([Values] bool withErrorBarriers) + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new($"INSERT INTO {table} (id) VALUES (9)"), + new("SELECT 1"), + new("INVALID SQL"), + new($"INSERT INTO {table} (id) VALUES (9)"), + new("SELECT 2") + }, + EnableErrorBarriers = withErrorBarriers + }; + + await using (var reader = await batch.ExecuteReaderAsync(Behavior)) + { + Assert.That(await reader.ReadAsync(), Is.True); + Assert.That(reader[0], Is.EqualTo(1)); + Assert.That(await reader.ReadAsync(), Is.False); + + Assert.That(async () => await reader.NextResultAsync(), Throws.Exception.TypeOf()); + + Assert.That(reader.State, Is.EqualTo(ReaderState.Consumed)); + Assert.That(await reader.ReadAsync(), Is.False); + Assert.That(await reader.NextResultAsync(), Is.False); + } + + Assert.That(await conn.ExecuteScalarAsync($"SELECT count(*) FROM {table}"), withErrorBarriers + ? Is.EqualTo(2) + : Is.EqualTo(0)); + } + + [Test] + public async Task Error_with_AppendErrorBarrier() + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new($"INSERT INTO {table} (id) VALUES (8)"), + new("INVALID SQL") { AppendErrorBarrier = true }, + new($"INSERT INTO {table} (id) VALUES (9)") + } + }; + + // A Sync is placed after the 2nd command (INVALID SQL), so the 1st command is rolled back but not the 3rd. + var exception = Assert.ThrowsAsync(async () => await batch.ExecuteReaderAsync(Behavior))!; + Assert.That(exception.BatchCommand, Is.SameAs(batch.BatchCommands[1])); + + Assert.That(await conn.ExecuteScalarAsync($"SELECT id FROM {table} ORDER BY id"), Is.EqualTo(9)); + } + + [Test] + public async Task Batch_with_terminating_error_barrier() + { + await using var conn = await OpenConnectionAsync(); + await using var _ = await CreateTempTable(conn, "id INT", out var table); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new($"INSERT INTO {table} (id) VALUES (8)"), + new($"INSERT INTO {table} (id) VALUES (9)") { AppendErrorBarrier = true } + } + }; + + Assert.That(await batch.ExecuteNonQueryAsync(), Is.EqualTo(2)); + } + + [Test] + public async Task Error_barriers_with_SchemaOnly() + { + await using var conn = await OpenConnectionAsync(); + + await using var batch = new NpgsqlBatch(conn) + { + BatchCommands = + { + new("SELECT 1"), + new("SELECT 'foo'") + }, + EnableErrorBarriers = true + }; + + await using var reader = await batch.ExecuteReaderAsync(CommandBehavior.SchemaOnly | Behavior); + + var columnSchema = await reader.GetColumnSchemaAsync(); + Assert.That(columnSchema[0].DataType, Is.SameAs(typeof(int))); + + Assert.That(await reader.NextResultAsync(), Is.True); + columnSchema = await reader.GetColumnSchemaAsync(); + Assert.That(columnSchema[0].DataType, Is.SameAs(typeof(string))); + } + + #endregion Error barriers + #region Miscellaneous [Test] From d658b834bfa47bd4cee3d79c435bead78866692a Mon Sep 17 00:00:00 2001 From: Shay Rojansky Date: Sun, 25 Sep 2022 14:06:09 +0200 Subject: [PATCH 2/3] WIP --- src/Npgsql/NpgsqlCommand.cs | 7 +++-- src/Npgsql/NpgsqlDataReader.cs | 53 +++++++++++++++------------------ test/Npgsql.Tests/BatchTests.cs | 7 +++-- 3 files changed, 32 insertions(+), 35 deletions(-) diff --git a/src/Npgsql/NpgsqlCommand.cs b/src/Npgsql/NpgsqlCommand.cs index 6cd50fe98d..573bafc233 100644 --- a/src/Npgsql/NpgsqlCommand.cs +++ b/src/Npgsql/NpgsqlCommand.cs @@ -963,12 +963,14 @@ internal Task Write(NpgsqlConnector connector, bool async, bool flush, Cancellat async Task WriteExecute(NpgsqlConnector connector, bool async, bool flush, CancellationToken cancellationToken) { + NpgsqlBatchCommand? batchCommand = null; + for (var i = 0; i < InternalBatchCommands.Count; i++) { // The following is only for deadlock avoidance when doing sync I/O (so never in multiplexing) ForceAsyncIfNecessary(ref async, i); - var batchCommand = InternalBatchCommands[i]; + batchCommand = InternalBatchCommands[i]; var pStatement = batchCommand.PreparedStatement; Debug.Assert(batchCommand.FinalCommandText is not null); @@ -1009,8 +1011,7 @@ await connector.WriteBind( pStatement.LastUsed = DateTime.UtcNow; } - if (InternalBatchCommands.Count == 0 || - !EnableErrorBarriers && InternalBatchCommands[InternalBatchCommands.Count - 1].AppendErrorBarrier != true) + if (batchCommand is null || !(batchCommand.AppendErrorBarrier ?? EnableErrorBarriers)) { await connector.WriteSync(async, cancellationToken); } diff --git a/src/Npgsql/NpgsqlDataReader.cs b/src/Npgsql/NpgsqlDataReader.cs index 83cc3dd504..64868483a6 100644 --- a/src/Npgsql/NpgsqlDataReader.cs +++ b/src/Npgsql/NpgsqlDataReader.cs @@ -531,11 +531,9 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo } // There are no more queries, we're done. Read the RFQ. - if (_statements.Count == 0 || - !Command.EnableErrorBarriers && _statements[_statements.Count - 1].AppendErrorBarrier != true) - { + if (_statements.Count == 0 || !(_statements[_statements.Count - 1].AppendErrorBarrier ?? Command.EnableErrorBarriers)) Expect(await Connector.ReadMessage(async), Connector); - } + State = ReaderState.Consumed; RowDescription = null; return false; @@ -567,35 +565,32 @@ async Task NextResult(bool async, bool isConsuming = false, CancellationTo statement.PreparedStatement!.AbortPrepare(); } - if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) - break; - } - - // In normal, non-isolated batching, we've consumed the result set and are done. - // However, if an isolated command was present after the error, we now have to consume the rest of the result set. - // Note that Consume calls NextResult (this method) recursively, the isConsuming flag tells us we're in this mode. - if (StatementIndex == _statements.Count) - { - State = ReaderState.Consumed; - } - else if (!isConsuming) - { - switch (State) + // In normal, non-isolated batching, we've consumed the result set and are done. + // However, if the command has error barrier, we now have to consume results from the commands after it (unless it's the + // last one). + // Note that Consume calls NextResult (this method) recursively, the isConsuming flag tells us we're in this mode. + if ((statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) && StatementIndex < _statements.Count - 1) { - case ReaderState.Consumed: - case ReaderState.Closed: - case ReaderState.Disposed: - // The exception may have caused the connector to break (e.g. I/O), and so the reader is already closed. - break; - default: - // We provide Consume with the first exception which we've just caught. - // If it encounters other exceptions while consuming the rest of the result set, it will raise an AggregateException, - // otherwise it will rethrow this first exception. - await Consume(async, firstException: e); - break; + if (isConsuming) + throw; + switch (State) + { + case ReaderState.Consumed: + case ReaderState.Closed: + case ReaderState.Disposed: + // The exception may have caused the connector to break (e.g. I/O), and so the reader is already closed. + break; + default: + // We provide Consume with the first exception which we've just caught. + // If it encounters other exceptions while consuming the rest of the result set, it will raise an AggregateException, + // otherwise it will rethrow this first exception. + await Consume(async, firstException: e); + break; // Never reached, Consume always throws above + } } } + State = ReaderState.Consumed; throw; } } diff --git a/test/Npgsql.Tests/BatchTests.cs b/test/Npgsql.Tests/BatchTests.cs index 816e81b775..23a84e89aa 100644 --- a/test/Npgsql.Tests/BatchTests.cs +++ b/test/Npgsql.Tests/BatchTests.cs @@ -519,7 +519,7 @@ public async Task Error_with_AppendErrorBarrier() } [Test] - public async Task Batch_with_terminating_error_barrier() + public async Task AppendErrorBarrier_on_last_command([Values] bool enabled) { await using var conn = await OpenConnectionAsync(); await using var _ = await CreateTempTable(conn, "id INT", out var table); @@ -529,8 +529,9 @@ public async Task Batch_with_terminating_error_barrier() BatchCommands = { new($"INSERT INTO {table} (id) VALUES (8)"), - new($"INSERT INTO {table} (id) VALUES (9)") { AppendErrorBarrier = true } - } + new($"INSERT INTO {table} (id) VALUES (9)") { AppendErrorBarrier = enabled } + }, + EnableErrorBarriers = true }; Assert.That(await batch.ExecuteNonQueryAsync(), Is.EqualTo(2)); From 464f48d8c4b6040668a3c3234ff8d30c5b4b5dc3 Mon Sep 17 00:00:00 2001 From: Shay Rojansky Date: Sun, 25 Sep 2022 14:43:27 +0200 Subject: [PATCH 3/3] Optimize Expect --- src/Npgsql/Util/PGUtil.cs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/Npgsql/Util/PGUtil.cs b/src/Npgsql/Util/PGUtil.cs index 97430badcd..1eef1e6838 100644 --- a/src/Npgsql/Util/PGUtil.cs +++ b/src/Npgsql/Util/PGUtil.cs @@ -1,6 +1,7 @@ using Npgsql.Internal; using System; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Reflection; using System.Runtime.CompilerServices; using System.Text; @@ -27,10 +28,18 @@ static Statics() [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static T Expect(IBackendMessage msg, NpgsqlConnector connector) - => msg is T t - ? t - : throw connector.Break( + { + if (msg is T t) + return t; + + Throw(msg, connector); + return default; + + [MethodImpl(MethodImplOptions.NoInlining), DoesNotReturn] + static void Throw(IBackendMessage msg, NpgsqlConnector connector) + => throw connector.Break( new NpgsqlException($"Received backend message {msg.Code} while expecting {typeof(T).Name}. Please file a bug.")); + } internal static DeferDisposable Defer(Action action) => new(action); internal static DeferDisposable Defer(Action action, T arg) => new(action, arg);