using System;

using System.Collections;

using System.Collections.Generic;

using System.Collections.ObjectModel;

using System.Data;

using System.Data.Common;

using System.Diagnostics;

using System.Diagnostics.CodeAnalysis;

using System.IO;

using System.Linq;

using System.Runtime.CompilerServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using Microsoft.Extensions.Logging;

using Npgsql.BackendMessages;

using Npgsql.Internal;

using Npgsql.Internal.TypeHandlers;

using Npgsql.Internal.TypeHandling;

using Npgsql.PostgresTypes;

using Npgsql.Schema;

using Npgsql.Util;

using NpgsqlTypes;

using static Npgsql.Util.Statics;

#pragma warning disable CA2222 // Do not decrease inherited member visibility

namespace Npgsql;

/// <summary>

/// Reads a forward-only stream of rows from a data source.

/// </summary>

#pragma warning disable CA1010

public sealed class NpgsqlDataReader : DbDataReader, IDbColumnSchemaGenerator

#pragma warning restore CA1010

{

internal NpgsqlCommand Command { get; private set; } = default!;

internal NpgsqlConnector Connector { get; }

NpgsqlConnection? _connection;

/// <summary>

/// The behavior of the command with which this reader was executed.

/// </summary>

CommandBehavior _behavior;

/// <summary>

/// In multiplexing, this is <see langword="null" /> as the sending is managed in the write multiplexing loop,

/// and does not need to be awaited by the reader.

/// </summary>

Task? _sendTask;

internal ReaderState State = ReaderState.Disposed;

internal NpgsqlReadBuffer Buffer = default!;

/// <summary>

/// Holds the list of statements being executed by this reader.

/// </summary>

List<NpgsqlBatchCommand> _statements = default!;

/// <summary>

/// The index of the current query resultset we're processing (within a multiquery)

/// </summary>

internal int StatementIndex { get; private set; }

/// <summary>

/// The number of columns in the current row

/// </summary>

int _numColumns;

/// <summary>

/// Records, for each column, its starting offset and length in the current row.

/// Used only in non-sequential mode.

/// </summary>

readonly List<(int Offset, int Length)> _columns = new();

/// <summary>

/// The index of the column that we're on, i.e. that has already been parsed, is

/// is memory and can be retrieved. Initialized to -1, which means we're on the column

/// count (which comes before the first column).

/// </summary>

int _column;

/// <summary>

/// For streaming types (e.g. bytea), holds the byte length of the column.

/// Does not include the length prefix.

/// </summary>

internal int ColumnLen;

internal int PosInColumn;

/// <summary>

/// The position in the buffer at which the current data row message ends.

/// Used only when the row is consumed non-sequentially.

/// </summary>

int _dataMsgEnd;

/// <summary>

/// Determines, if we can consume the row non-sequentially.

/// Mostly useful for a sequential mode, when the row is already in the buffer.

/// Should always be true for the non-sequential mode.

/// </summary>

bool _canConsumeRowNonSequentially;

int _charPos;

/// <summary>

/// The RowDescription message for the current resultset being processed

/// </summary>

internal RowDescriptionMessage? RowDescription;

ulong? _recordsAffected;

/// <summary>

/// Whether the current result set has rows

/// </summary>

bool _hasRows;

/// <summary>

/// Is raised whenever Close() is called.

/// </summary>

public event EventHandler? ReaderClosed;

bool _isSchemaOnly;

bool _isSequential;

/// <summary>

/// A stream that has been opened on a column.

/// </summary>

NpgsqlReadBuffer.ColumnStream? _columnStream;

/// <summary>

/// Used for internal temporary purposes

/// </summary>

char[]? _tempCharBuf;

/// <summary>

/// Used to keep track of every unique row this reader object ever traverses.

/// This is used to detect whether nested DbDataReaders are still valid.

/// </summary>

internal ulong UniqueRowId;

internal NpgsqlNestedDataReader? CachedFreeNestedDataReader;

static readonly ILogger Logger = NpgsqlLoggingConfiguration.CommandLogger;

internal NpgsqlDataReader(NpgsqlConnector connector)

{

Connector = connector;

}

internal void Init(

NpgsqlCommand command, CommandBehavior behavior, List<NpgsqlBatchCommand> statements, Task? sendTask = null)

{

Command = command;

_connection = command.Connection;

_behavior = behavior;

_isSchemaOnly = _behavior.HasFlag(CommandBehavior.SchemaOnly);

_isSequential = _behavior.HasFlag(CommandBehavior.SequentialAccess);

_statements = statements;

StatementIndex = -1;

_sendTask = sendTask;

State = ReaderState.BetweenResults;

_recordsAffected = null;

}

#region Read

/// <summary>

/// Advances the reader to the next record in a result set.

/// </summary>

/// <returns><b>true</b> if there are more rows; otherwise <b>false</b>.</returns>

/// <remarks>

/// The default position of a data reader is before the first record. Therefore, you must call Read to begin accessing data.

/// </remarks>

public override bool Read()

{

CheckClosedOrDisposed();

UniqueRowId++;

var fastRead = TryFastRead();

return fastRead.HasValue

? fastRead.Value

: Read(false).GetAwaiter().GetResult();

}

/// <summary>

/// This is the asynchronous version of <see cref="Read()"/>

/// </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> ReadAsync(CancellationToken cancellationToken)

{

CheckClosedOrDisposed();

UniqueRowId++;

var fastRead = TryFastRead();

if (fastRead.HasValue)

return fastRead.Value ? PGUtil.TrueTask : PGUtil.FalseTask;

using (NoSynchronizationContextScope.Enter())

return Read(true, cancellationToken);

}

bool? TryFastRead()

{

// This is an optimized execution path that avoids calling any async methods for the (usual)

// case where the next row (or CommandComplete) is already in memory.

if (_behavior.HasFlag(CommandBehavior.SingleRow))

return null;

switch (State)

{

case ReaderState.BeforeResult:

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

State = ReaderState.InResult;

return true;

case ReaderState.InResult:

if (!_canConsumeRowNonSequentially)

return null;

// We get here, if we're in a non-sequential mode (or the row is already in the buffer)

ConsumeRowNonSequential();

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

case ReaderState.Disposed:

return false;

}

var readBuf = Connector.ReadBuffer;

if (readBuf.ReadBytesLeft < 5)

return null;

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

var len = readBuf.ReadInt32() - 4; // Transmitted length includes itself

if (messageCode != BackendMessageCode.DataRow || readBuf.ReadBytesLeft < len)

{

readBuf.ReadPosition -= 5;

return null;

}

var msg = Connector.ParseServerMessage(readBuf, messageCode, len, false)!;

Debug.Assert(msg.Code == BackendMessageCode.DataRow);

ProcessMessage(msg);

return true;

}

async Task<bool> Read(bool async, CancellationToken cancellationToken = default)

{

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);

if (_behavior.HasFlag(CommandBehavior.SingleRow))

{

// TODO: See optimization proposal in #410

await Consume(async);

return false;

}

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

case ReaderState.Disposed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

var msg2 = await ReadMessage(async);

ProcessMessage(msg2);

return msg2.Code == BackendMessageCode.DataRow;

}

catch

{

State = ReaderState.Consumed;

throw;

}

finally

{

registration.Dispose();

}

}

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);

if (msg.Code == BackendMessageCode.DataRow)

{

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

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

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() => (_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)

{

using (NoSynchronizationContextScope.Enter())

return _isSchemaOnly

? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken)

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

}

/// <summary>

/// Internal implementation of NextResult

/// </summary>

[MethodImpl(MethodImplOptions.AggressiveInlining)]

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

{

CheckClosedOrDisposed();

IBackendMessage msg;

Debug.Assert(!_isSchemaOnly);

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

try

{

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

switch (State)

{

case ReaderState.BeforeResult:

case ReaderState.InResult:

await ConsumeRow(async);

while (true)

{

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

switch (completedMsg.Code)

{

case BackendMessageCode.CommandComplete:

case BackendMessageCode.EmptyQueryResponse:

ProcessMessage(completedMsg);

break;

default:

continue;

}

break;

}

break;

case ReaderState.BetweenResults:

break;

case ReaderState.Consumed:

case ReaderState.Closed:

case ReaderState.Disposed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

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

_hasRows = false;

if (_behavior.HasFlag(CommandBehavior.SingleResult) && StatementIndex == 0 && !isConsuming)

{

await Consume(async);

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

{

var statement = _statements[StatementIndex];

if (statement.TryGetPrepared(out var preparedStatement))

{

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

RowDescription = preparedStatement.Description;

}

else // Non-prepared/preparing flow

{

var pStatement = statement.PreparedStatement;

if (pStatement != null)

{

Debug.Assert(!pStatement.IsPrepared);

if (pStatement.StatementBeingReplaced != null)

{

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

pStatement.StatementBeingReplaced.CompleteUnprepare();

pStatement.StatementBeingReplaced = null;

}

}

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

if (statement.IsPreparing)

{

pStatement!.State = PreparedState.Prepared;

Connector.PreparedStatementManager.NumPrepared++;

statement.IsPreparing = false;

}

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

msg = await Connector.ReadMessage(async);

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 => pStatement == null

? (RowDescriptionMessage)msg

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

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

};

}

if (RowDescription == null)

{

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

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

msg = await ReadMessage(async);

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."));

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

ProcessMessage(msg);

continue;

}

if (!Command.IsWrappedByBatch && StatementIndex == 0 && Command.Parameters.HasOutputParameters)

{

// If output parameters are present and this is the first row of the first 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);

ProcessMessage(msg);

if (msg.Code == BackendMessageCode.DataRow)

PopulateOutputParameters();

}

else

{

msg = await ReadMessage(async);

ProcessMessage(msg);

}

switch (msg.Code)

{

case BackendMessageCode.DataRow:

case BackendMessageCode.CommandComplete:

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

return true;

}

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

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

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

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

{

Command.IsCached = 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;

}

}

void PopulateOutputParameters()

{

// The first row in a stored procedure command that has output parameters needs to be traversed twice -

// once for populating the output parameters and once for the actual result set traversal. So in this

// case we can't be sequential.

Debug.Assert(Command.Parameters.Any(p => p.IsOutputDirection));

Debug.Assert(StatementIndex == 0);

Debug.Assert(RowDescription != null);

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

var currentPosition = Buffer.ReadPosition;

// Temporarily set our state to InResult to allow us to read the values

State = ReaderState.InResult;

var pending = new Queue<object>();

var taken = new List<NpgsqlParameter>();

for (var i = 0; i < FieldCount; i++)

{

if (Command.Parameters.TryGetValue(GetName(i), out var p) && p.IsOutputDirection)

{

p.Value = GetValue(i);

taken.Add(p);

}

else

pending.Enqueue(GetValue(i));

}

// Not sure where this odd behavior comes from: all output parameters which did not get matched by

// name now get populated with column values which weren't matched. Keeping this for backwards compat,

// opened #2252 for investigation.

foreach (var p in Command.Parameters.Where(p => p.IsOutputDirection && !taken.Contains(p)))

{

if (pending.Count == 0)

break;

p.Value = pending.Dequeue();

}

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

Buffer.ReadPosition = currentPosition; // Restore position

_column = -1;

ColumnLen = -1;

PosInColumn = 0;

}

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

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

try

{

switch (State)

{

case ReaderState.BeforeResult:

case ReaderState.InResult:

case ReaderState.BetweenResults:

break;

case ReaderState.Consumed:

case ReaderState.Closed:

case ReaderState.Disposed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

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

{

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

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

var msg = await Connector.ReadMessage(async);

switch (msg.Code)

{

case BackendMessageCode.NoData:

RowDescription = _statements[StatementIndex].Description = null;

break;

case BackendMessageCode.RowDescription:

// We have a resultset

RowDescription = _statements[StatementIndex].Description = (RowDescriptionMessage)msg;

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

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

}

// Found a resultset

if (RowDescription != null)

return true;

}

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

if (!_statements.All(s => s.IsPrepared))

{

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

RowDescription = null;

}

return false;

}

catch (Exception e)

{

State = ReaderState.Consumed;

// Reference the triggering statement from the exception (for batching)

if (e is PostgresException postgresException &&

Command.IsWrappedByBatch &&

StatementIndex >= 0 && StatementIndex < _statements.Count)

{

postgresException.BatchCommand = _statements[StatementIndex];

}

throw;

}

}

#endregion

#region ProcessMessage

internal void ProcessMessage(IBackendMessage msg)

{

switch (msg.Code)

{

case BackendMessageCode.DataRow:

ProcessDataRowMessage((DataRowMessage)msg);

return;

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:

if (!_recordsAffected.HasValue)

_recordsAffected = 0;

_recordsAffected += completed.Rows;

break;

}

_statements[StatementIndex].ApplyCommandComplete(completed);

goto case BackendMessageCode.EmptyQueryResponse;

case BackendMessageCode.EmptyQueryResponse:

State = ReaderState.BetweenResults;

return;

case BackendMessageCode.ReadyForQuery:

State = ReaderState.Consumed;

return;

default:

throw new Exception("Received unexpected backend message of type " + msg.Code);

}

}

void ProcessDataRowMessage(DataRowMessage msg)

{

Connector.State = ConnectorState.Fetching;

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

Buffer = Connector.ReadBuffer;

_hasRows = true;

_column = -1;

ColumnLen = -1;

PosInColumn = 0;

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

_numColumns = Buffer.ReadInt16();

Debug.Assert(_numColumns == RowDescription!.Count,

$"Row's number of columns ({_numColumns}) differs from the row description's ({RowDescription.Count})");

_dataMsgEnd = Buffer.ReadPosition + msg.Length - 2;

_canConsumeRowNonSequentially = Buffer.ReadBytesLeft >= msg.Length - 2;

if (!_isSequential)

{

Debug.Assert(_canConsumeRowNonSequentially);

// Initialize our columns array with the offset and length of the first column

_columns.Clear();

var len = Buffer.ReadInt32();

_columns.Add((Buffer.ReadPosition, len));

}

switch (State)

{

case ReaderState.BetweenResults:

State = ReaderState.BeforeResult;

break;

case ReaderState.BeforeResult:

State = ReaderState.InResult;

break;

case ReaderState.InResult:

break;

default:

throw Connector.UnexpectedMessageReceived(BackendMessageCode.DataRow);

}

}

#endregion

void Cancel() => Connector.PerformPostgresCancellation();

/// <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 == ReaderState.Closed || State == 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 => _statements.AsReadOnly();

/// <summary>

/// Gets a value that indicates whether this DbDataReader contains one or more rows.

/// </summary>

public override bool HasRows

=> State switch

{

ReaderState.Closed => throw new InvalidOperationException("Invalid attempt to call HasRows when reader is closed."),

ReaderState.Disposed => throw new ObjectDisposedException(nameof(NpgsqlDataReader)),

_ => _hasRows

};

/// <summary>

/// Indicates whether the reader is currently positioned on a row, i.e. whether reading a

/// column is possible.

/// This property is different from <see cref="HasRows"/> in that <see cref="HasRows"/> will

/// return true even if attempting to read a column will fail, e.g. before <see cref="Read()"/>

/// has been called

/// </summary>

public bool IsOnRow => State == ReaderState.InResult;

/// <summary>

/// Gets the name of the column, given the zero-based column ordinal.

/// </summary>

/// <param name="ordinal">The zero-based column ordinal.</param>

/// <returns>The name of the specified column.</returns>

public override string GetName(int ordinal) => GetField(ordinal).Name;

/// <summary>

/// Gets the number of columns in the current row.

/// </summary>

public override int FieldCount

{

get

{

CheckClosedOrDisposed();

return RowDescription?.Count ?? 0;

}

}

#region Cleanup / Dispose

/// <summary>

/// Consumes all result sets for this reader, leaving the connector ready for sending and processing further

/// queries

/// </summary>

async Task Consume(bool async)

{

// 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)) {}

}

/// <summary>

/// Releases the resources used by the <see cref="NpgsqlDataReader"/>.

/// </summary>

protected override void Dispose(bool disposing)

{

try

{

Close(connectionClosing: false, async: false, isDisposing: true).GetAwaiter().GetResult();

}

catch (Exception e)

{

Logger.LogError(e, "Exception caught while disposing a reader", Connector.Id);

if (e is not PostgresException)

State = ReaderState.Disposed;

}

finally

{

Command.TraceCommandStop();

}

}

/// <summary>

/// Releases the resources used by the <see cref="NpgsqlDataReader"/>.

/// </summary>

#if NETSTANDARD2_0

public ValueTask DisposeAsync()

#else

public override ValueTask DisposeAsync()

#endif

{

using (NoSynchronizationContextScope.Enter())

return DisposeAsyncCore();

[MethodImpl(MethodImplOptions.AggressiveInlining)]

async ValueTask DisposeAsyncCore()

{

try

{

await Close(connectionClosing: false, async: true, isDisposing: true);

}

catch (Exception e)

{

Logger.LogError(e, "Exception caught while disposing a reader", Connector.Id);

if (e is not PostgresException)

State = ReaderState.Disposed;

}

finally

{

Command.TraceCommandStop();

}

}

}

/// <summary>

/// Closes the <see cref="NpgsqlDataReader"/> reader, allowing a new command to be executed.

/// </summary>

public override void Close() => Close(connectionClosing: false, async: false, isDisposing: false).GetAwaiter().GetResult();

/// <summary>

/// Closes the <see cref="NpgsqlDataReader"/> reader, allowing a new command to be executed.

/// </summary>

#if NETSTANDARD2_0

public Task CloseAsync()

#else

public override Task CloseAsync()

#endif

{

using (NoSynchronizationContextScope.Enter())

return Close(connectionClosing: false, async: true, isDisposing: false);

}

internal async Task Close(bool connectionClosing, bool async, bool isDisposing)

{

if (State == ReaderState.Closed || State == ReaderState.Disposed)

{

if (isDisposing)

State = ReaderState.Disposed;

return;

}

// Whenever a connector is broken, it also closes the current reader.

Connector.CurrentReader = null;

switch (Connector.State)

{

case ConnectorState.Ready:

case ConnectorState.Fetching:

case ConnectorState.Executing:

case ConnectorState.Connecting:

if (State != ReaderState.Consumed)

{

try

{

await Consume(async);

}

catch (Exception ex) when (

ex is OperationCanceledException ||

ex is NpgsqlException && ex.InnerException is TimeoutException)

{

// Timeout/cancellation - completely normal, consume has basically completed.

}

catch (PostgresException)

{

// In the case of a PostgresException, the connection is fine and consume has basically completed.

// Defer throwing the exception until Cleanup is complete.

await Cleanup(async, connectionClosing, isDisposing);

throw;

}

catch

{

Debug.Assert(Connector.IsBroken);

throw;

}

}

break;

case ConnectorState.Closed:

case ConnectorState.Broken:

break;

case ConnectorState.Waiting:

case ConnectorState.Copy:

case ConnectorState.Replication:

Debug.Fail("Bad connector state when closing reader: " + Connector.State);

break;

default:

throw new ArgumentOutOfRangeException();

}

await Cleanup(async, connectionClosing, isDisposing);

}

internal async Task Cleanup(bool async, bool connectionClosing = false, bool isDisposing = false)