#region Fields

NpgsqlTransaction? _transaction;

readonly NpgsqlConnector? _connector;

/// <summary>

/// If this command is (explicitly) prepared, references the connector on which the preparation happened.

/// Used to detect when the connector was changed (i.e. connection open/close), meaning that the command

/// is no longer prepared.

/// </summary>

NpgsqlConnector? _connectorPreparedOn;

string _commandText;

CommandBehavior _behavior;

int? _timeout;

internal NpgsqlParameterCollection? _parameters;

/// <summary>

/// Whether this <see cref="NpgsqlCommand" /> is wrapped by an <see cref="NpgsqlBatch" />.

/// </summary>

internal bool IsWrappedByBatch { get; }

internal List<NpgsqlBatchCommand> InternalBatchCommands { get; }

Activity? CurrentActivity;

/// <summary>

/// Returns details about each statement that this command has executed.

/// Is only populated when an Execute* method is called.

/// </summary>

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

public IReadOnlyList<NpgsqlBatchCommand> Statements => InternalBatchCommands.AsReadOnly();

UpdateRowSource _updateRowSource = UpdateRowSource.Both;

bool IsExplicitlyPrepared => _connectorPreparedOn != null;

/// <summary>

/// Whether this command is cached by <see cref="NpgsqlConnection" /> and returned by <see cref="NpgsqlConnection.CreateCommand" />.

/// </summary>

internal bool IsCacheable { get; set; }

#if DEBUG

internal static bool EnableSqlRewriting;

internal static bool EnableStoredProcedureCompatMode;

#else

internal static readonly bool EnableSqlRewriting;

internal static readonly bool EnableStoredProcedureCompatMode;

#endif

internal bool EnableErrorBarriers { get; set; }

static readonly TaskScheduler ConstrainedConcurrencyScheduler =

new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, Math.Max(1, Environment.ProcessorCount / 2)).ConcurrentScheduler;

#endregion Fields

#region Constants

internal const int DefaultTimeout = 30;

#endregion

#region Constructors

static NpgsqlCommand()

{

EnableSqlRewriting = !AppContext.TryGetSwitch("Npgsql.EnableSqlRewriting", out var enabled) || enabled;

EnableStoredProcedureCompatMode = AppContext.TryGetSwitch("Npgsql.EnableStoredProcedureCompatMode", out enabled) && enabled;

}

/// <summary>

/// Initializes a new instance of the <see cref="NpgsqlCommand"/> class.

/// </summary>

public NpgsqlCommand() : this(null, null, null) {}

/// <summary>

/// Initializes a new instance of the <see cref="NpgsqlCommand"/> class with the text of the query.

/// </summary>

/// <param name="cmdText">The text of the query.</param>

// ReSharper disable once IntroduceOptionalParameters.Global

public NpgsqlCommand(string? cmdText) : this(cmdText, null, null) {}

/// <summary>

/// Initializes a new instance of the <see cref="NpgsqlCommand"/> class with the text of the query and a

/// <see cref="NpgsqlConnection"/>.

/// </summary>

/// <param name="cmdText">The text of the query.</param>

/// <param name="connection">A <see cref="NpgsqlConnection"/> that represents the connection to a PostgreSQL server.</param>

// ReSharper disable once IntroduceOptionalParameters.Global

public NpgsqlCommand(string? cmdText, NpgsqlConnection? connection)

{

GC.SuppressFinalize(this);

InternalBatchCommands = new List<NpgsqlBatchCommand>(1);

_commandText = cmdText ?? string.Empty;

InternalConnection = connection;

CommandType = CommandType.Text;

}

/// <summary>

/// Initializes a new instance of the <see cref="NpgsqlCommand"/> class with the text of the query, a

/// <see cref="NpgsqlConnection"/>, and the <see cref="NpgsqlTransaction"/>.

/// </summary>

/// <param name="cmdText">The text of the query.</param>

/// <param name="connection">A <see cref="NpgsqlConnection"/> that represents the connection to a PostgreSQL server.</param>

/// <param name="transaction">The <see cref="NpgsqlTransaction"/> in which the <see cref="NpgsqlCommand"/> executes.</param>

public NpgsqlCommand(string? cmdText, NpgsqlConnection? connection, NpgsqlTransaction? transaction)

: this(cmdText, connection)

=> Transaction = transaction;

/// <summary>

/// Used when this <see cref="NpgsqlCommand"/> instance is wrapped inside an <see cref="NpgsqlBatch"/>.

/// </summary>

internal NpgsqlCommand(int batchCommandCapacity, NpgsqlConnection? connection = null)

{

GC.SuppressFinalize(this);

InternalBatchCommands = new List<NpgsqlBatchCommand>(batchCommandCapacity);

InternalConnection = connection;

CommandType = CommandType.Text;

IsWrappedByBatch = true;

// These can/should never be used in this mode

_commandText = null!;

_parameters = null!;

}

internal NpgsqlCommand(string? cmdText, NpgsqlConnector connector) : this(cmdText)

=> _connector = connector;

/// <summary>

/// Used when this <see cref="NpgsqlCommand"/> instance is wrapped inside an <see cref="NpgsqlBatch"/>.

/// </summary>

internal NpgsqlCommand(NpgsqlConnector connector, int batchCommandCapacity)

: this(batchCommandCapacity)

=> _connector = connector;

internal static NpgsqlCommand CreateCachedCommand(NpgsqlConnection connection)

=> new(null, connection) { IsCacheable = true };

#endregion Constructors

#region Public properties

/// <summary>

/// Gets or sets the SQL statement or function (stored procedure) to execute at the data source.

/// </summary>

/// <value>The SQL statement or function (stored procedure) to execute. The default is an empty string.</value>

[AllowNull, DefaultValue("")]

[Category("Data")]

public override string CommandText

{

get => _commandText;

set

{

Debug.Assert(!IsWrappedByBatch);

if (State != CommandState.Idle)

ThrowHelper.ThrowInvalidOperationException("An open data reader exists for this command.");

_commandText = value ?? string.Empty;

ResetPreparation();

// TODO: Technically should do this also if the parameter list (or type) changes

}

}

/// <summary>

/// Gets or sets the wait time (in seconds) before terminating the attempt to execute a command and generating an error.

/// </summary>

/// <value>The time (in seconds) to wait for the command to execute. The default value is 30 seconds.</value>

[DefaultValue(DefaultTimeout)]

public override int CommandTimeout

{

get => _timeout ?? (InternalConnection?.CommandTimeout ?? DefaultTimeout);

set

{

if (value < 0) {

throw new ArgumentOutOfRangeException(nameof(value), value, "CommandTimeout can't be less than zero.");

}

_timeout = value;

}

}

/// <summary>

/// Gets or sets a value indicating how the <see cref="NpgsqlCommand.CommandText"/> property is to be interpreted.

/// </summary>

/// <value>

/// One of the <see cref="System.Data.CommandType"/> values. The default is <see cref="System.Data.CommandType.Text"/>.

/// </value>

[DefaultValue(CommandType.Text)]

[Category("Data")]

public override CommandType CommandType { get; set; }

internal NpgsqlConnection? InternalConnection { get; private set; }

/// <summary>

/// DB connection.

/// </summary>

protected override DbConnection? DbConnection

{

get => InternalConnection;

set

{

if (InternalConnection == value)

return;

InternalConnection = State == CommandState.Idle

? (NpgsqlConnection?)value

: throw new InvalidOperationException("An open data reader exists for this command.");

Transaction = null;

}

}

/// <summary>

/// Gets or sets the <see cref="NpgsqlConnection"/> used by this instance of the <see cref="NpgsqlCommand"/>.

/// </summary>

/// <value>The connection to a data source. The default value is <see langword="null"/>.</value>

[DefaultValue(null)]

[Category("Behavior")]

public new NpgsqlConnection? Connection

{

get => (NpgsqlConnection?)DbConnection;

set => DbConnection = value;

}

/// <summary>

/// Design time visible.

/// </summary>

public override bool DesignTimeVisible { get; set; }

/// <summary>

/// Gets or sets how command results are applied to the DataRow when used by the

/// DbDataAdapter.Update(DataSet) method.

/// </summary>

/// <value>One of the <see cref="System.Data.UpdateRowSource"/> values.</value>

[Category("Behavior"), DefaultValue(UpdateRowSource.Both)]

public override UpdateRowSource UpdatedRowSource

{

get => _updateRowSource;

set

{

switch (value)

{

// validate value (required based on base type contract)

case UpdateRowSource.None:

case UpdateRowSource.OutputParameters:

case UpdateRowSource.FirstReturnedRecord:

case UpdateRowSource.Both:

_updateRowSource = value;

break;

default:

throw new ArgumentOutOfRangeException();

}

}

}

/// <summary>

/// Returns whether this query will execute as a prepared (compiled) query.

/// </summary>

public bool IsPrepared

{

get

{

return _connectorPreparedOn == (InternalConnection?.Connector ?? _connector) && AllPrepared();

bool AllPrepared()

{

if (InternalBatchCommands.Count is 0)

return false;

foreach (var s in InternalBatchCommands)

if (s.PreparedStatement is null || !s.PreparedStatement.IsPrepared)

return false;

return true;

}

}

}

#endregion Public properties

#region Known/unknown Result Types Management

/// <summary>

/// Marks all of the query's result columns as either known or unknown.

/// Unknown result columns are requested from PostgreSQL in text format, and Npgsql makes no

/// attempt to parse them. They will be accessible as strings only.

/// </summary>

public bool AllResultTypesAreUnknown

{

get => _allResultTypesAreUnknown;

set

{

// TODO: Check that this isn't modified after calling prepare

_unknownResultTypeList = null;

_allResultTypesAreUnknown = value;

}

}

bool _allResultTypesAreUnknown;

/// <summary>

/// Marks the query's result columns as known or unknown, on a column-by-column basis.

/// Unknown result columns are requested from PostgreSQL in text format, and Npgsql makes no

/// attempt to parse them. They will be accessible as strings only.

/// </summary>

/// <remarks>

/// If the query includes several queries (e.g. SELECT 1; SELECT 2), this will only apply to the first

/// one. The rest of the queries will be fetched and parsed as usual.

///

/// The array size must correspond exactly to the number of result columns the query returns, or an

/// error will be raised.

/// </remarks>

public bool[]? UnknownResultTypeList

{

get => _unknownResultTypeList;

set

{

// TODO: Check that this isn't modified after calling prepare

_allResultTypesAreUnknown = false;

_unknownResultTypeList = value;

}

}

bool[]? _unknownResultTypeList;

#endregion

#region State management

volatile int _state;

/// <summary>

/// The current state of the command

/// </summary>

internal CommandState State

{

get => (CommandState)_state;

set

{

var newState = (int)value;

if (newState == _state)

return;

_state = newState;

}

}

internal void ResetPreparation() => _connectorPreparedOn = null;

#endregion State management

#region Parameters

/// <summary>

/// Creates a new instance of an <see cref="System.Data.Common.DbParameter"/> object.

/// </summary>

/// <returns>A <see cref="System.Data.Common.DbParameter"/> object.</returns>

protected override DbParameter CreateDbParameter() => CreateParameter();

/// <summary>

/// Creates a new instance of a <see cref="NpgsqlParameter"/> object.

/// </summary>

/// <returns>An <see cref="NpgsqlParameter"/> object.</returns>

public new NpgsqlParameter CreateParameter() => new();

/// <summary>

/// DB parameter collection.

/// </summary>

protected override DbParameterCollection DbParameterCollection => Parameters;

/// <summary>

/// Gets the <see cref="NpgsqlParameterCollection"/>.

/// </summary>

/// <value>The parameters of the SQL statement or function (stored procedure). The default is an empty collection.</value>

public new NpgsqlParameterCollection Parameters => _parameters ??= new();

#endregion

#region DeriveParameters

const string DeriveParametersForFunctionQuery = @"

internal void DeriveParameters()

{

var conn = CheckAndGetConnection();

Debug.Assert(conn is not null);

if (string.IsNullOrEmpty(CommandText))

throw new InvalidOperationException("CommandText property has not been initialized");

using var _ = conn.StartTemporaryBindingScope(out var connector);

foreach (var s in InternalBatchCommands)

if (s.PreparedStatement?.IsExplicit == true)

throw new NpgsqlException("Deriving parameters isn't supported for commands that are already prepared.");

// Here we unprepare statements that possibly are auto-prepared

Unprepare();

Parameters.Clear();

switch (CommandType)

{

case CommandType.Text:

DeriveParametersForQuery(connector);

break;

case CommandType.StoredProcedure:

DeriveParametersForFunction();

break;

default:

throw new NotSupportedException("Cannot derive parameters for CommandType " + CommandType);

}

}

void DeriveParametersForFunction()

{

using var c = new NpgsqlCommand(DeriveParametersForFunctionQuery, InternalConnection);

c.Parameters.Add(new NpgsqlParameter("proname", NpgsqlDbType.Text));

c.Parameters[0].Value = CommandText;

string[]? names = null;

uint[]? types = null;

char[]? modes = null;

using (var rdr = c.ExecuteReader(CommandBehavior.SingleRow | CommandBehavior.SingleResult))

{

if (rdr.Read())

{

if (!rdr.IsDBNull(0))

names = rdr.GetFieldValue<string[]>(0);

if (!rdr.IsDBNull(2))

types = rdr.GetFieldValue<uint[]>(2);

if (!rdr.IsDBNull(3))

modes = rdr.GetFieldValue<char[]>(3);

if (types == null)

{

if (rdr.IsDBNull(1) || rdr.GetFieldValue<uint[]>(1).Length == 0)

return; // Parameter-less function

types = rdr.GetFieldValue<uint[]>(1);

}

}

else

throw new InvalidOperationException($"{CommandText} does not exist in pg_proc");

}

var serializerOptions = c.InternalConnection!.Connector!.SerializerOptions;

for (var i = 0; i < types.Length; i++)

{

var param = new NpgsqlParameter();

var postgresType = serializerOptions.DatabaseInfo.GetPostgresType(types[i]);

var npgsqlDbType = postgresType.DataTypeName.ToNpgsqlDbType();

param.DataTypeName = postgresType.DisplayName;

param.PostgresType = postgresType;

if (npgsqlDbType.HasValue)

param.NpgsqlDbType = npgsqlDbType.Value;

if (names != null && i < names.Length)

param.ParameterName = names[i];

else

param.ParameterName = "parameter" + (i + 1);

if (modes == null) // All params are IN, or server < 8.1.0 (and only IN is supported)

param.Direction = ParameterDirection.Input;

else

{

param.Direction = modes[i] switch

{

'i' => ParameterDirection.Input,

'o' => ParameterDirection.Output,

't' => ParameterDirection.Output,

'b' => ParameterDirection.InputOutput,

'v' => throw new NotSupportedException("Cannot derive function parameter of type VARIADIC"),

_ => throw new ArgumentOutOfRangeException("Unknown code in proargmodes while deriving: " + modes[i])

};

}

Parameters.Add(param);

}

}

void DeriveParametersForQuery(NpgsqlConnector connector)

{

using (connector.StartUserAction())

{

LogMessages.DerivingParameters(connector.CommandLogger, CommandText, connector.Id);

if (IsWrappedByBatch)

foreach (var batchCommand in InternalBatchCommands)

connector.SqlQueryParser.ParseRawQuery(batchCommand, connector.UseConformingStrings, deriveParameters: true);

else

connector.SqlQueryParser.ParseRawQuery(this, connector.UseConformingStrings, deriveParameters: true);

var sendTask = SendDeriveParameters(connector, false);

if (sendTask.IsFaulted)

sendTask.GetAwaiter().GetResult();

try

{

foreach (var batchCommand in InternalBatchCommands)

{

Expect<ParseCompleteMessage>(

connector.ReadMessage(async: false).GetAwaiter().GetResult(), connector);

var paramTypeOIDs = Expect<ParameterDescriptionMessage>(

connector.ReadMessage(async: false).GetAwaiter().GetResult(), connector).TypeOIDs;

if (batchCommand.PositionalParameters.Count != paramTypeOIDs.Count)

{

connector.SkipUntil(BackendMessageCode.ReadyForQuery);

Parameters.Clear();

throw new NpgsqlException(

"There was a mismatch in the number of derived parameters between the Npgsql SQL parser and the PostgreSQL parser. Please report this as bug to the Npgsql developers (https://github.com/npgsql/npgsql/issues).");

}

for (var i = 0; i < paramTypeOIDs.Count; i++)

{

try

{

var param = batchCommand.PositionalParameters[i];

var paramOid = paramTypeOIDs[i];

var postgresType = connector.SerializerOptions.DatabaseInfo.GetPostgresType(paramOid);

// We want to keep any domain types visible on the parameter, it will internally do a representational lookup again if necessary.

var npgsqlDbType = postgresType.GetRepresentationalType().DataTypeName.ToNpgsqlDbType();

if (param.NpgsqlDbType != NpgsqlDbType.Unknown && param.NpgsqlDbType != npgsqlDbType)

throw new NpgsqlException(

"The backend parser inferred different types for parameters with the same name. Please try explicit casting within your SQL statement or batch or use different placeholder names.");

param.DataTypeName = postgresType.DisplayName;

param.PostgresType = postgresType;

if (npgsqlDbType.HasValue)

param.NpgsqlDbType = npgsqlDbType.Value;

}

catch

{

connector.SkipUntil(BackendMessageCode.ReadyForQuery);

Parameters.Clear();

throw;

}

}

var msg = connector.ReadMessage(async: false).GetAwaiter().GetResult();

switch (msg.Code)

{

case BackendMessageCode.RowDescription:

case BackendMessageCode.NoData:

break;

default:

throw connector.UnexpectedMessageReceived(msg.Code);

}

}

Expect<ReadyForQueryMessage>(connector.ReadMessage(async: false).GetAwaiter().GetResult(), connector);

}

finally

{

try

{

// Make sure sendTask is complete so we don't race against asynchronous flush

sendTask.GetAwaiter().GetResult();

}

catch

{

// ignored

}

}

}

}

#endregion

#region Prepare

/// <summary>

/// Creates a server-side prepared statement on the PostgreSQL server.

/// This will make repeated future executions of this command much faster.

/// </summary>

public override void Prepare() => Prepare(false).GetAwaiter().GetResult();

/// <summary>

/// Creates a server-side prepared statement on the PostgreSQL server.

/// This will make repeated future executions of this command much faster.

/// </summary>

/// <param name="cancellationToken">

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

/// </param>

public override Task PrepareAsync(CancellationToken cancellationToken = default)

=> Prepare(async: true, cancellationToken);

Task Prepare(bool async, CancellationToken cancellationToken = default)

{

var connection = CheckAndGetConnection();

Debug.Assert(connection is not null);

if (connection.Settings.Multiplexing)

throw new NotSupportedException("Explicit preparation not supported with multiplexing");

var connector = connection.Connector!;

var logger = connector.CommandLogger;

var needToPrepare = false;

if (IsWrappedByBatch)

{

foreach (var batchCommand in InternalBatchCommands)

{

batchCommand._parameters?.ProcessParameters(connector.SerializerOptions, validateValues: false, CommandType);

ProcessRawQuery(connector.SqlQueryParser, connector.UseConformingStrings, batchCommand);

needToPrepare = batchCommand.ExplicitPrepare(connector) || needToPrepare;

}

if (logger.IsEnabled(LogLevel.Debug) && needToPrepare)

LogMessages.PreparingCommandExplicitly(logger, string.Join("; ", CommandTexts()), connector.Id);

IEnumerable<string> CommandTexts()

{

foreach (var c in InternalBatchCommands)

yield return c.CommandText;

}

}

else

{

_parameters?.ProcessParameters(connector.SerializerOptions, validateValues: false, CommandType);

ProcessRawQuery(connector.SqlQueryParser, connector.UseConformingStrings, batchCommand: null);

foreach (var batchCommand in InternalBatchCommands)

needToPrepare = batchCommand.ExplicitPrepare(connector) || needToPrepare;

if (logger.IsEnabled(LogLevel.Debug) && needToPrepare)

LogMessages.PreparingCommandExplicitly(logger, CommandText, connector.Id);

}

_connectorPreparedOn = connector;

// It's possible the command was already prepared, or that persistent prepared statements were found for

// all statements. Nothing to do here, move along.

return needToPrepare

? PrepareLong(this, async, connector, cancellationToken)

: Task.CompletedTask;

static async Task PrepareLong(NpgsqlCommand command, bool async, NpgsqlConnector connector, CancellationToken cancellationToken)

{

try

{

using (connector.StartUserAction(cancellationToken))

{

var sendTask = command.SendPrepare(connector, async, CancellationToken.None);

if (sendTask.IsFaulted)

sendTask.GetAwaiter().GetResult();

try

{

// Loop over statements, skipping those that are already prepared (because they were persisted)

var isFirst = true;

foreach (var batchCommand in command.InternalBatchCommands)

{

if (!batchCommand.IsPreparing)

continue;

var pStatement = batchCommand.PreparedStatement!;

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

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

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

switch (msg.Code)

{

case BackendMessageCode.RowDescription:

// Clone the RowDescription for use with the prepared statement (the one we have is reused

// by the connection)

var description = ((RowDescriptionMessage)msg).Clone();

command.FixupRowDescription(description, isFirst);

batchCommand.Description = description;

break;

case BackendMessageCode.NoData:

batchCommand.Description = null;

break;

default:

throw connector.UnexpectedMessageReceived(msg.Code);

}

pStatement.State = PreparedState.Prepared;

connector.PreparedStatementManager.NumPrepared++;

batchCommand.IsPreparing = false;

isFirst = false;

}

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

}

finally

{

try

{

// Make sure sendTask is complete so we don't race against asynchronous flush

if (async)

await sendTask.ConfigureAwait(false);

else

sendTask.GetAwaiter().GetResult();

}

catch

{

// ignored

}

}

}

LogMessages.CommandPreparedExplicitly(connector.CommandLogger, connector.Id);

}

catch

{

// The statements weren't prepared successfully, update the bookkeeping for them

foreach (var batchCommand in command.InternalBatchCommands)

{

if (batchCommand.IsPreparing)

{

batchCommand.IsPreparing = false;

batchCommand.PreparedStatement!.AbortPrepare();

}

}

throw;

}