{

get => _isRowBuffered;

set => _isRowBuffered = value;

{

Connector = connector;

_commandLogger = connector.CommandLogger;

NpgsqlCommand command,

CommandBehavior behavior,

List<NpgsqlBatchCommand> statements,

long startTimestamp = 0,

Task? sendTask = null)

{

Debug.Assert(ColumnInfoCache is null);

Command = command;

_connection = command.InternalConnection;

_behavior = behavior;

_isSchemaOnly = _behavior.HasFlag(CommandBehavior.SchemaOnly);

_isSequential = _behavior.HasFlag(CommandBehavior.SequentialAccess);

_statements = statements;

StatementIndex = -1;

_sendTask = sendTask;

State = ReaderState.BetweenResults;

_recordsAffected = null;

_startTimestamp = startTimestamp;

{

ThrowIfClosedOrDisposed();

return TryRead()?.Result ?? Read(false).GetAwaiter().GetResult();

{

ThrowIfClosedOrDisposed();

return TryRead() ?? Read(async: true, cancellationToken);

{

switch (State)

{

case ReaderState.BeforeResult:

// First Read() after NextResult. Data row has already been processed.

State = ReaderState.InResult;

return TrueTask;

case ReaderState.InResult:

break;

default:

return FalseTask;

}

// We have a special case path for SingleRow.

if (_behavior.HasFlag(CommandBehavior.SingleRow) || !_isRowBuffered)

return null;

ConsumeBufferedRow();

const int headerSize = sizeof(byte) + sizeof(int);

var buffer = Buffer;

var readPosition = buffer.ReadPosition;

var bytesLeft = buffer.FilledBytes - readPosition;

if (bytesLeft < headerSize)

return null;

var messageCode = (BackendMessageCode)buffer.ReadByte();

var len = buffer.ReadInt32() - sizeof(int); // Transmitted length includes itself

var isDataRow = messageCode is BackendMessageCode.DataRow;

// sizeof(short) is for the number of columns

var sufficientBytes = isDataRow && _isSequential ? headerSize + sizeof(short) : headerSize + len;

if (bytesLeft < sufficientBytes

|| !isDataRow && (_statements[StatementIndex].AppendErrorBarrier ?? Command.EnableErrorBarriers)

// Could be an error, let main read handle it.

|| Connector.ParseResultSetMessage(buffer, messageCode, len) is not { } msg)

{

buffer.ReadPosition = readPosition;

return null;

}

ProcessMessage(msg);

return isDataRow ? TrueTask : FalseTask;

{

using var registration = Connector.StartNestedCancellableOperation(cancellationToken);

try

{

switch (State)

{

case ReaderState.BeforeResult:

// First Read() after NextResult. Data row has already been processed.

State = ReaderState.InResult;

return true;

case ReaderState.InResult:

await ConsumeRow(async).ConfigureAwait(false);

if (_behavior.HasFlag(CommandBehavior.SingleRow))

{

// TODO: See optimization proposal in #410

await Consume(async).ConfigureAwait(false);

return false;

}

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

case ReaderState.Disposed:

return false;

default:

ThrowHelper.ThrowArgumentOutOfRangeException();

return false;

}

var msg = await ReadMessage(async).ConfigureAwait(false);

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<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

return false;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

}

catch

{

// Break may have progressed the reader already.

if (State is not ReaderState.Closed)

State = ReaderState.Consumed;

throw;

}

}

ValueTask<IBackendMessage> ReadMessage(bool async)

{

return _isSequential ? ReadMessageSequential(Connector, async) : Connector.ReadMessage(async);

static async ValueTask<IBackendMessage> ReadMessageSequential(NpgsqlConnector connector, bool async)

{

var msg = await connector.ReadMessage(async, DataRowLoadingMode.Sequential).ConfigureAwait(false);

if (msg.Code == BackendMessageCode.DataRow)

{

// Make sure that the datarow's column count is already buffered

await connector.ReadBuffer.Ensure(2, async).ConfigureAwait(false);

return msg;

}

return msg;

}

}

#endregion

#region NextResult

/// <summary>

/// Advances the reader to the next result when reading the results of a batch of statements.

/// </summary>

/// <returns></returns>

public override bool NextResult()

{

ThrowIfClosedOrDisposed();

return (_isSchemaOnly ? NextResultSchemaOnly(false) : NextResult(false))

.GetAwaiter().GetResult();

}

/// <summary>

/// This is the asynchronous version of NextResult.

/// </summary>

/// <param name="cancellationToken">

/// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>.

/// </param>

/// <returns>A task representing the asynchronous operation.</returns>

public override Task<bool> NextResultAsync(CancellationToken cancellationToken)

{

ThrowIfClosedOrDisposed();

return _isSchemaOnly

? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken)

: NextResult(async: true, cancellationToken: cancellationToken);

}

/// <summary>

/// Internal implementation of NextResult

/// </summary>

async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationToken cancellationToken = default)

{

Debug.Assert(!_isSchemaOnly);

if (State is ReaderState.Consumed)

return false;

try

{

using var registration = isConsuming ? default : Connector.StartNestedCancellableOperation(cancellationToken);

// If we're in the middle of a resultset, consume it

if (State is ReaderState.BeforeResult or ReaderState.InResult)

await ConsumeResultSet(async).ConfigureAwait(false);

Debug.Assert(State is ReaderState.BetweenResults);

_hasRows = false;

var statements = _statements;

var statementIndex = StatementIndex;

if (statementIndex >= 0)

{

if (RowDescription is { } description && statements[statementIndex].IsPrepared && ColumnInfoCache is { } cache)

description.SetColumnInfoCache(new(cache, 0, ColumnCount));

if (statementIndex is 0 && _behavior.HasFlag(CommandBehavior.SingleResult) && !isConsuming)

{

await Consume(async).ConfigureAwait(false);

return false;

}

}

// We are now at the end of the previous result set. Read up to the next result set, if any.

// Non-prepared statements receive ParseComplete, BindComplete, DescriptionRow/NoData,

// prepared statements receive only BindComplete

for (statementIndex = ++StatementIndex; statementIndex < statements.Count; statementIndex = ++StatementIndex)

{

var statement = statements[statementIndex];

IBackendMessage msg;

if (statement.TryGetPrepared(out var preparedStatement))

{

Expect<BindCompleteMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

RowDescription = preparedStatement.Description;

}

else // Non-prepared/preparing flow

{

preparedStatement = statement.PreparedStatement;

if (preparedStatement != null)

{

Debug.Assert(!preparedStatement.IsPrepared);

if (preparedStatement.StatementBeingReplaced != null)

{

Expect<CloseCompletedMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

preparedStatement.StatementBeingReplaced.CompleteUnprepare();

preparedStatement.StatementBeingReplaced = null;

}

}

Expect<ParseCompleteMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

if (statement.IsPreparing)

{

preparedStatement!.State = PreparedState.Prepared;

Connector.PreparedStatementManager.NumPrepared++;

statement.IsPreparing = false;

}

Expect<BindCompleteMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

msg = await Connector.ReadMessage(async).ConfigureAwait(false);

RowDescription = statement.Description = msg.Code switch

{

BackendMessageCode.NoData => null,

// RowDescription messages are cached on the connector, but if we're auto-preparing, we need to

// clone our own copy which will last beyond the lifetime of this invocation.

BackendMessageCode.RowDescription => preparedStatement == null

? (RowDescriptionMessage)msg

: ((RowDescriptionMessage)msg).Clone(),

_ => throw Connector.UnexpectedMessageReceived(msg.Code)

};

}

if (RowDescription is not null)

{

if (ColumnInfoCache?.Length >= ColumnCount)

Array.Clear(ColumnInfoCache, 0, ColumnCount);

else

{

if (ColumnInfoCache is { } cache)

ArrayPool<ColumnInfo>.Shared.Return(cache, clearArray: true);

ColumnInfoCache = ArrayPool<ColumnInfo>.Shared.Rent(ColumnCount);

}

if (statement.IsPrepared)

RowDescription.LoadColumnInfoCache(Connector.SerializerOptions, ColumnInfoCache);

}

else

{

// Statement did not generate a resultset (e.g. INSERT)

// Read and process its completion message and move on to the next statement

// No need to read sequentially as it's not a DataRow

msg = await Connector.ReadMessage(async).ConfigureAwait(false);

switch (msg.Code)

{

case BackendMessageCode.CommandComplete:

case BackendMessageCode.EmptyQueryResponse:

break;

case BackendMessageCode.CopyInResponse:

throw Connector.Break(new NotSupportedException(

"COPY isn't supported in regular command execution - see https://www.npgsql.org/doc/copy.html for documentation on COPY with Npgsql. " +

"If you are trying to execute a SQL script created by pg_dump, pass the '--inserts' switch to disable generating COPY statements."));

case BackendMessageCode.CopyOutResponse:

throw Connector.Break(new NotSupportedException(

"COPY isn't supported in regular command execution - see https://www.npgsql.org/doc/copy.html for documentation on COPY with Npgsql."));

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

ProcessMessage(msg);

if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers)

Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

continue;

}

if ((Command.WrappingBatch is not null || StatementIndex is 0) && Command.InternalBatchCommands[StatementIndex] is { HasOutputParameters: true } command)

{

// If output parameters are present and this is the first row of the resultset,

// we must always read it in non-sequential mode because it will be traversed twice (once

// here for the parameters, then as a regular row).

msg = await Connector.ReadMessage(async, dataRowLoadingMode: DataRowLoadingMode.NonSequential).ConfigureAwait(false);

ProcessMessage(msg);

if (msg.Code == BackendMessageCode.DataRow)

{

Debug.Assert(RowDescription != null);

Debug.Assert(State == ReaderState.BeforeResult);

try

{

// Temporarily set our state to InResult and non-sequential to allow us to read the values, and in any order.

var isSequential = _isSequential;

var currentPosition = Buffer.ReadPosition;

State = ReaderState.InResult;

_isSequential = false;

try

{

command.PopulateOutputParameters(this, _commandLogger);

// On success we want to revert any row and column state for the user to be able to read the same row again.

if (async)

await PgReader.CommitAsync().ConfigureAwait(false);

else

PgReader.Commit();

State = ReaderState.BeforeResult; // Set the state back

Buffer.ReadPosition = currentPosition; // Restore position

_column = -1;

}

finally

{

// To be on the safe side we always revert this CommandBehavior state change, including on failure.

_isSequential = isSequential;

}

}

catch (Exception e)

{

// TODO: ideally we should flow down to global exception filter and consume there

await Consume(async, firstException: e).ConfigureAwait(false);

throw;

}

}

}

else

{

msg = await ReadMessage(async).ConfigureAwait(false);

ProcessMessage(msg);

}

switch (msg.Code)

{

case BackendMessageCode.DataRow:

Connector.State = ConnectorState.Fetching;

return true;

case BackendMessageCode.CommandComplete:

if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers)

Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

return true;

default:

Connector.UnexpectedMessageReceived(msg.Code);

break;

}

}

// There are no more queries, we're done. Read the RFQ.

if (_statements.Count is 0 || !(_statements[^1].AppendErrorBarrier ?? Command.EnableErrorBarriers))

Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

State = ReaderState.Consumed;

RowDescription = null;

return false;

}

catch (Exception e)

{

if (e is PostgresException postgresException && StatementIndex >= 0 && StatementIndex < _statements.Count)

{

var statement = _statements[StatementIndex];

// Reference the triggering statement from the exception

if (Connector.Settings.IncludeFailedBatchedCommand)

postgresException.BatchCommand = statement;

// 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.

// TODO: we probably should do than even if it's not PostgresException (error from PopulateOutputParameters)

Command.IsCacheable = false;

// If the schema of a table changes after a statement is prepared on that table, PostgreSQL errors with

// 0A000: cached plan must not change result type. 0A000 seems like a non-specific code, but it's very unlikely the

// statement would successfully execute anyway, so invalidate the prepared statement.

if (postgresException.SqlState == PostgresErrorCodes.FeatureNotSupported &&

statement.PreparedStatement is { } preparedStatement)

{

preparedStatement.State = PreparedState.Invalidated;

Command.ResetPreparation();

}

}

// 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];

if (statement.IsPreparing)

{

statement.IsPreparing = false;

statement.PreparedStatement!.AbortPrepare();

}

// 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.

// TODO: We might as well call Consume on every command (even the last one) to make sure we do read every single message until RFQ

// in case we get an exception in the middle of NextResult

if ((statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) && StatementIndex < _statements.Count - 1)

{

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).ConfigureAwait(false);

break; // Never reached, Consume always throws above

}

}

}

// Break may have progressed the reader already.

if (State is not ReaderState.Closed)

State = ReaderState.Consumed;

throw;

}

async ValueTask ConsumeResultSet(bool async)

{

await ConsumeRow(async).ConfigureAwait(false);

while (true)

{

var completedMsg = await Connector.ReadMessage(async, DataRowLoadingMode.Skip).ConfigureAwait(false);

switch (completedMsg.Code)

{

case BackendMessageCode.CommandComplete:

case BackendMessageCode.EmptyQueryResponse:

ProcessMessage(completedMsg);

var statement = _statements[StatementIndex];

if (statement.IsPrepared && ColumnInfoCache is not null)

RowDescription!.SetColumnInfoCache(new(ColumnInfoCache, 0, ColumnCount));

if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers)

Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

break;

default:

// TODO if we hit an ErrorResponse here (PG doesn't do this *today*) we should probably throw.

continue;

}

break;

}

}

}

/// <summary>

/// Note that in SchemaOnly mode there are no resultsets, and we read nothing from the backend (all

/// RowDescriptions have already been processed and are available)

/// </summary>

async Task<bool> NextResultSchemaOnly(bool async, bool isConsuming = false, CancellationToken cancellationToken = default)

{

Debug.Assert(_isSchemaOnly);

if (State is ReaderState.Consumed)

return false;

using var registration = isConsuming ? default : Connector.StartNestedCancellableOperation(cancellationToken);

try

{

for (StatementIndex++; StatementIndex < _statements.Count; StatementIndex++)

{

var statement = _statements[StatementIndex];

if (statement.TryGetPrepared(out var preparedStatement))

{

// Row descriptions have already been populated in the statement objects at the

// Prepare phase

RowDescription = preparedStatement.Description;

}

else

{

var pStatement = statement.PreparedStatement;

if (pStatement != null)

{

Debug.Assert(!pStatement.IsPrepared);

if (pStatement.StatementBeingReplaced != null)

{

Expect<CloseCompletedMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

pStatement.StatementBeingReplaced.CompleteUnprepare();

pStatement.StatementBeingReplaced = null;

}

}

Expect<ParseCompleteMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

if (statement.IsPreparing)

{

pStatement!.State = PreparedState.Prepared;

Connector.PreparedStatementManager.NumPrepared++;

statement.IsPreparing = false;

}

Expect<ParameterDescriptionMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

var msg = await Connector.ReadMessage(async).ConfigureAwait(false);

switch (msg.Code)

{

case BackendMessageCode.NoData:

RowDescription = _statements[StatementIndex].Description = null;

break;

case BackendMessageCode.RowDescription:

// We have a resultset

// RowDescription messages are cached on the connector, but if we're auto-preparing, we need to

// clone our own copy which will last beyond the lifetime of this invocation.

RowDescription = _statements[StatementIndex].Description = preparedStatement == null

? (RowDescriptionMessage)msg

: ((RowDescriptionMessage)msg).Clone();

Command.FixupRowDescription(RowDescription, StatementIndex == 0);

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

var forall = true;

for (var i = StatementIndex + 1; i < _statements.Count; i++)

if (!_statements[i].IsPrepared)

{

forall = false;

break;

}

// There are no more queries, we're done. Read to the RFQ.

if (forall)

Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async).ConfigureAwait(false), Connector);

}

// Found a resultset

if (RowDescription is not null)

return true;

}

State = ReaderState.Consumed;

RowDescription = null;

return false;

}

catch (Exception e)

{

// Break may have progressed the reader already.

if (State is not ReaderState.Closed)

State = ReaderState.Consumed;

// Reference the triggering statement from the exception

if (e is PostgresException postgresException && StatementIndex >= 0 && StatementIndex < _statements.Count)

{

// Reference the triggering statement from the exception

if (Connector.Settings.IncludeFailedBatchedCommand)

postgresException.BatchCommand = _statements[StatementIndex];

// 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.

Command.IsCacheable = false;

}

// 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 (; StatementIndex < _statements.Count; StatementIndex++)

{

var statement = _statements[StatementIndex];

if (statement.IsPreparing)

{

statement.IsPreparing = false;

statement.PreparedStatement!.AbortPrepare();

}

}

throw;

}

}

#endregion

#region ProcessMessage

internal void ProcessMessage(IBackendMessage msg)

{

if (msg.Code is not BackendMessageCode.DataRow)

{

HandleUncommon(msg);

return;

}

var dataRow = (DataRowMessage)msg;

// The connector's buffer can actually change between DataRows:

// If a large DataRow exceeding the connector's current read buffer arrives, and we're

// reading in non-sequential mode, a new oversize buffer is allocated. We thus have to

// recapture the connector's buffer on each new DataRow.

// Note that this can happen even in sequential mode, if the row description message is big

// (see #2003)

if (!ReferenceEquals(Buffer, Connector.ReadBuffer))

{

Buffer = Connector.ReadBuffer;

Buffer.PgReader.StreamCanSeek = !_isSequential;

}

// We assume that the row's number of columns is identical to the description's

var numColumns = Buffer.ReadInt16();

if (ColumnCount != numColumns)

ThrowHelper.ThrowArgumentException($"Row's number of columns ({numColumns}) differs from the row description's ({ColumnCount})");

var readPosition = Buffer.ReadPosition;

var msgRemainder = dataRow.Length - sizeof(short);

_dataMsgEnd = readPosition + msgRemainder;

_columnsStartPos = readPosition;

_isRowBuffered = msgRemainder <= Buffer.FilledBytes - readPosition;

Debug.Assert(_isRowBuffered || _isSequential);

_column = -1;

if (_columns.Count > 0)

_columns.Clear();

switch (State)

{

case ReaderState.BetweenResults:

_hasRows = true;

State = ReaderState.BeforeResult;

break;

case ReaderState.BeforeResult:

State = ReaderState.InResult;

break;

case ReaderState.InResult:

break;

default:

Connector.UnexpectedMessageReceived(BackendMessageCode.DataRow);

break;

}

[MethodImpl(MethodImplOptions.NoInlining)]

void HandleUncommon(IBackendMessage msg)

{

switch (msg.Code)

{

case BackendMessageCode.CommandComplete:

var completed = (CommandCompleteMessage)msg;

switch (completed.StatementType)

{

case StatementType.Update:

case StatementType.Insert:

case StatementType.Delete:

case StatementType.Copy:

case StatementType.Move:

case StatementType.Merge:

_recordsAffected ??= 0;

_recordsAffected += completed.Rows;

break;

}

_statements[StatementIndex].ApplyCommandComplete(completed);

State = ReaderState.BetweenResults;

break;

case BackendMessageCode.EmptyQueryResponse:

State = ReaderState.BetweenResults;

break;

default:

Connector.UnexpectedMessageReceived(msg.Code);

break;

}

}

}

#endregion

/// <summary>

/// Gets a value indicating the depth of nesting for the current row. Always returns zero.

/// </summary>

public override int Depth => 0;

/// <summary>

/// Gets a value indicating whether the data reader is closed.

/// </summary>

public override bool IsClosed => State is ReaderState.Closed or ReaderState.Disposed;

/// <summary>

/// Gets the number of rows changed, inserted, or deleted by execution of the SQL statement.

/// </summary>

/// <value>

/// The number of rows changed, inserted, or deleted. -1 for SELECT statements; 0 if no rows were affected or the statement failed.

/// </value>

public override int RecordsAffected

=> !_recordsAffected.HasValue

? -1

: _recordsAffected > int.MaxValue

? throw new OverflowException(

$"The number of records affected exceeds int.MaxValue. Use {nameof(Rows)}.")

: (int)_recordsAffected;

/// <summary>

/// Gets the number of rows changed, inserted, or deleted by execution of the SQL statement.

/// </summary>

/// <value>

/// The number of rows changed, inserted, or deleted. 0 for SELECT statements, if no rows were affected or the statement failed.

/// </value>

public ulong Rows => _recordsAffected ?? 0;

/// <summary>

/// Returns details about each statement that this reader will or has executed.

/// </summary>

/// <remarks>

/// Note that some fields (i.e. rows and oid) are only populated as the reader

/// traverses the result.

///

/// For commands with multiple queries, this exposes the number of rows affected on

/// a statement-by-statement basis, unlike <see cref="NpgsqlDataReader.RecordsAffected"/>

/// which exposes an aggregation across all statements.

/// </remarks>

[Obsolete("Use the new DbBatch API")]

public IReadOnlyList<NpgsqlBatchCommand> Statements

{

get

{

ThrowIfClosedOrDisposed();

return _statements.AsReadOnly();

}

{

get

{

ThrowIfClosedOrDisposed();

return _hasRows;

}

{

get

{

ThrowIfClosedOrDisposed();

return State is ReaderState.InResult;

}

{

get

{

ThrowIfClosedOrDisposed();

return RowDescription?.Count ?? 0;

}