#region License

#endregion

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Data;

using System.Data.Common;

using System.Diagnostics;

using System.Linq;

using System.Runtime.CompilerServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using System.Globalization;

using System.Net.Sockets;

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.Logging;

using NpgsqlTypes;

namespace Npgsql

/// <summary>

/// Represents a SQL statement or function (stored procedure) to execute

/// against a PostgreSQL database. This class cannot be inherited.

/// </summary>

#if NETSTANDARD1_3

public sealed class NpgsqlCommand : DbCommand

#else

// ReSharper disable once RedundantNameQualifier

[System.ComponentModel.DesignerCategory("")]

public sealed class NpgsqlCommand : DbCommand, ICloneable

#endif

{

#region Fields

[CanBeNull]

NpgsqlConnection _connection;

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

[CanBeNull]

NpgsqlConnector _connectorPreparedOn;

NpgsqlTransaction _transaction;

string _commandText;

int? _timeout;

readonly NpgsqlParameterCollection _parameters;

readonly List<NpgsqlStatement> _statements;

/// <summary>

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

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

/// </summary>

public IReadOnlyList<NpgsqlStatement> Statements => _statements.AsReadOnly();

UpdateRowSource _updateRowSource = UpdateRowSource.Both;

bool IsExplicitlyPrepared => _connectorPreparedOn != null;

static readonly SingleThreadSynchronizationContext SingleThreadSynchronizationContext = new SingleThreadSynchronizationContext("NpgsqlRemainingAsyncSendWorker");

static readonly NpgsqlLogger Log = NpgsqlLogManager.GetCurrentClassLogger();

#endregion Fields

#region Constants

internal const int DefaultTimeout = 30;

#endregion

#region Constructors

/// <summary>

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

/// </summary>

public NpgsqlCommand() : this(string.Empty, null, null) {}

/// <summary>

/// Initializes a new instance of the <see cref="NpgsqlCommand">NpgsqlCommand</see> 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">NpgsqlCommand</see> class with the text of the query and a <see cref="NpgsqlConnection">NpgsqlConnection</see>.

/// </summary>

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

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

// ReSharper disable once IntroduceOptionalParameters.Global

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

/// <summary>

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

/// </summary>

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

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

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

public NpgsqlCommand(string cmdText, [CanBeNull] NpgsqlConnection connection, [CanBeNull] NpgsqlTransaction transaction)

}

#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 Transact-SQL statement or stored procedure to execute. The default is an empty string.</value>

[DefaultValue("")]

[Category("Data")]

public override string CommandText

}

/// <summary>

/// Gets or sets the wait time 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

}

/// <summary>

/// Gets or sets a value indicating how the

/// <see cref="NpgsqlCommand.CommandText">CommandText</see> property is to be interpreted.

/// </summary>

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

[DefaultValue(CommandType.Text)]

[Category("Data")]

public override CommandType CommandType { get; set; }

/// <summary>

/// DB connection.

/// </summary>

protected override DbConnection DbConnection

}

/// <summary>

/// Gets or sets the <see cref="NpgsqlConnection">NpgsqlConnection</see>

/// used by this instance of the <see cref="NpgsqlCommand">NpgsqlCommand</see>.

/// </summary>

/// <value>The connection to a data source. The default value is a null reference.</value>

[DefaultValue(null)]

[Category("Behavior")]

[CanBeNull]

public new NpgsqlConnection Connection

}

/// <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">UpdateRowSource</see> values.</value>

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

public override UpdateRowSource UpdatedRowSource

}

/// <summary>

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

/// </summary>

public bool IsPrepared =>

_connectorPreparedOn == Connection?.Connector &&

_statements.Any() && _statements.All(s => s.PreparedStatement?.IsPrepared == 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 results column are requested them from PostgreSQL in text format, and Npgsql makes no

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

/// </summary>

public bool AllResultTypesAreUnknown

}

bool _allResultTypesAreUnknown;

/// <summary>

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

/// Unknown results column are requested them 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

}

bool[] _unknownResultTypeList;

#endregion

#region Result Types Management

/// <summary>

/// Marks result types to be used when using GetValue on a data reader, on a column-by-column basis.

/// Used for Entity Framework 5-6 compability.

/// Only primitive numerical types and DateTimeOffset are supported.

/// Set the whole array or just a value to null to use default type.

/// </summary>

internal Type[] ObjectResultTypes { get; set; }

#endregion

#region State management

int _state;

/// <summary>

/// Gets the current state of the connector

/// </summary>

internal CommandState State

}

void ResetExplicitPreparation() => _connectorPreparedOn = null;

#endregion State management

#region Parameters

/// <summary>

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

/// </summary>

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

protected override DbParameter CreateDbParameter()

{

return CreateParameter();

}

/// <summary>

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

/// </summary>

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

public new NpgsqlParameter CreateParameter()

{

return new NpgsqlParameter();

}

/// <summary>

/// DB parameter collection.

/// </summary>

protected override DbParameterCollection DbParameterCollection => Parameters;

/// <summary>

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

/// </summary>

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

public new NpgsqlParameterCollection Parameters => _parameters;

#endregion

#region Prepare

/// <summary>

/// Creates a prepared version of the command on a PostgreSQL server.

/// </summary>

public override void Prepare()

}

/// <summary>

/// Unprepares a command, closing server-side statements associated with it.

/// Note that this only affects commands explicitly prepared with <see cref="Prepare"/>, not

/// automatically prepared statements.

/// </summary>

public void Unprepare()

}

#endregion Prepare

#region Query analysis

void ProcessRawQuery()

}

#endregion

#region Execute

void ValidateParameters()

}

async ValueTask<NpgsqlDataReader> Execute(CommandBehavior behavior, bool async, CancellationToken cancellationToken)

{

ValidateParameters();

if ((behavior & CommandBehavior.SequentialAccess) != 0 && Parameters.HasOutputParameters)

throw new NotSupportedException("Output parameters aren't supported with SequentialAccess");

var connector = Connection.Connector;

Debug.Assert(connector != null);

if (IsExplicitlyPrepared)

{

Debug.Assert(_connectorPreparedOn != null);

if (_connectorPreparedOn != Connection.Connector)

{

// The command was prepared, but since then the connector has changed. Detach all prepared statements.

foreach (var s in _statements)

s.PreparedStatement = null;

ResetExplicitPreparation();

ProcessRawQuery();

}

}

else

{

ProcessRawQuery();

}

State = CommandState.InProgress;

try

{

if (Log.IsEnabled(NpgsqlLogLevel.Debug))

LogCommand();

Task sendTask;

// If a cancellation is in progress, wait for it to "complete" before proceeding (#615)

lock (connector.CancelLock) { }

connector.UserTimeout = CommandTimeout * 1000;

if ((behavior & CommandBehavior.SchemaOnly) == 0)

{

if (connector.Settings.MaxAutoPrepare > 0)

{

foreach (var statement in _statements)

{

// If this statement isn't prepared, see if it gets implicitly prepared.

// Note that this may return null (not enough usages for automatic preparation).

if (!statement.IsPrepared)

statement.PreparedStatement =

connector.PreparedStatementManager.TryGetAutoPrepared(statement);

if (statement.PreparedStatement != null)

statement.PreparedStatement.LastUsed = DateTime.UtcNow;

}

_connectorPreparedOn = connector;

}

// We do not wait for the entire send to complete before proceeding to reading -

// the sending continues in parallel with the user's reading. Waiting for the

// entire send to complete would trigger a deadlock for multistatement commands,

// where PostgreSQL sends large results for the first statement, while we're sending large

// parameter data for the second. See #641.

// Instead, all sends for non-first statements and for non-first buffers are performed

// asynchronously (even if the user requested sync), in a special synchronization context

// to prevents a dependency on the thread pool (which would also trigger deadlocks).

// The WriteBuffer notifies this command when the first buffer flush occurs, so that the

// send functions can switch to the special async mode when needed.

sendTask = SendExecute(async, cancellationToken);

}

else

{

sendTask = SendExecuteSchemaOnly(async, cancellationToken);

}

// The following is a hack. It raises an exception if one was thrown in the first phases

// of the send (i.e. in parts of the send that executed synchronously). Exceptions may

// still happen later and aren't properly handled. See #1323.

if (sendTask.IsFaulted)

sendTask.GetAwaiter().GetResult();

//var reader = new NpgsqlDataReader(this, behavior, _statements, sendTask);

var reader = (behavior & CommandBehavior.SequentialAccess) == 0

? (NpgsqlDataReader)new NpgsqlDefaultDataReader(this, behavior, _statements, sendTask)

: new NpgsqlSequentialDataReader(this, behavior, _statements, sendTask);

connector.CurrentReader = reader;

if (async)

await reader.NextResultAsync(cancellationToken);

else

reader.NextResult();

return reader;

}

catch

{

State = CommandState.Idle;

throw;

}

}

#endregion

#region Message Creation / Population

internal bool FlushOccurred { get; set; }

void BeginSend()

{

Debug.Assert(Connection?.Connector != null);

Connection.Connector.WriteBuffer.CurrentCommand = this;

FlushOccurred = false;

}

void CleanupSend()

{

// ReSharper disable once ConditionIsAlwaysTrueOrFalse

if (SynchronizationContext.Current != null) // Check first because SetSynchronizationContext allocates

SynchronizationContext.SetSynchronizationContext(null);

}

async Task SendExecute(bool async, CancellationToken cancellationToken)

{

BeginSend();

var connector = Connection.Connector;

Debug.Assert(connector != null);

var buf = connector.WriteBuffer;

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

{

if (!async && FlushOccurred && i > 0)

{

// We're synchronously sending the non-first statement in a batch and a flush

// has already occured. Switch to async. See long comment in Execute() above.

async = true;

SynchronizationContext.SetSynchronizationContext(SingleThreadSynchronizationContext);

}

var statement = _statements[i];

var pStatement = statement.PreparedStatement;

if (pStatement == null || pStatement.State == PreparedState.ToBePrepared)

{

if (pStatement?.StatementBeingReplaced != null)

{

// We have a prepared statement that replaces an existing statement - close the latter first.

await connector.CloseMessage

.Populate(StatementOrPortal.Statement, pStatement.StatementBeingReplaced.Name)

.Write(buf, async);

}

await connector.ParseMessage

.Populate(statement.SQL, statement.StatementName, statement.InputParameters, connector.TypeMapper)

.Write(buf, async);

}

var bind = connector.BindMessage;

bind.Populate(statement.InputParameters, "", statement.StatementName);

if (AllResultTypesAreUnknown)

bind.AllResultTypesAreUnknown = AllResultTypesAreUnknown;

else if (i == 0 && UnknownResultTypeList != null)

bind.UnknownResultTypeList = UnknownResultTypeList;

await connector.BindMessage.Write(buf, async);

if (pStatement == null || pStatement.State == PreparedState.ToBePrepared)

{

await connector.DescribeMessage

.Populate(StatementOrPortal.Portal)

.Write(buf, async);

if (statement.PreparedStatement != null)

statement.PreparedStatement.State = PreparedState.BeingPrepared;

}

await ExecuteMessage.DefaultExecute.Write(buf, async);

}

await SyncMessage.Instance.Write(buf, async);

await buf.Flush(async);

CleanupSend();

}

async Task SendExecuteSchemaOnly(bool async, CancellationToken cancellationToken)

{

BeginSend();

var connector = Connection.Connector;

Debug.Assert(connector != null);

var wroteSomething = false;

var buf = connector.WriteBuffer;

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

{

if (!async && FlushOccurred && i > 0)

{

// We're synchronously sending the non-first statement in a batch and a flush

// has already occured. Switch to async. See long comment in Execute() above.

async = true;

SynchronizationContext.SetSynchronizationContext(SingleThreadSynchronizationContext);

}

var statement = _statements[i];

if (statement.PreparedStatement?.State == PreparedState.Prepared)

continue; // Prepared, we already have the RowDescription

Debug.Assert(statement.PreparedStatement == null);

await connector.ParseMessage

.Populate(statement.SQL, "", statement.InputParameters, connector.TypeMapper)

.Write(buf, async);

await connector.DescribeMessage

.Populate(StatementOrPortal.Statement, statement.StatementName)

.Write(buf, async);

wroteSomething = true;

}

if (wroteSomething)

{

await SyncMessage.Instance.Write(buf, async);

await buf.Flush(async);

}

CleanupSend();

}

async Task SendPrepare(bool async, CancellationToken cancellationToken)

{

BeginSend();

var connector = Connection.Connector;

Debug.Assert(connector != null);

var buf = connector.WriteBuffer;

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

{

if (!async && FlushOccurred && i > 0)

{

// We're synchronously sending the non-first statement in a batch and a flush

// has already occured. Switch to async. See long comment in Execute() above.

async = true;

SynchronizationContext.SetSynchronizationContext(SingleThreadSynchronizationContext);

}

var statement = _statements[i];

var pStatement = statement.PreparedStatement;

// A statement may be already prepared, already in preparation (i.e. same statement twice

// in the same command), or we can't prepare (overloaded SQL)

if (pStatement?.State != PreparedState.ToBePrepared)

continue;

var statementToClose = pStatement.StatementBeingReplaced;

if (statementToClose != null)

{

// We have a prepared statement that replaces an existing statement - close the latter first.

await connector.CloseMessage

.Populate(StatementOrPortal.Statement, statementToClose.Name)

.Write(buf, async);

}

await connector.ParseMessage

.Populate(statement.SQL, pStatement.Name, statement.InputParameters, connector.TypeMapper)

.Write(buf, async);

await connector.DescribeMessage

.Populate(StatementOrPortal.Statement, pStatement.Name)

.Write(buf, async);

pStatement.State = PreparedState.BeingPrepared;

}

await SyncMessage.Instance.Write(buf, async);

await buf.Flush(async);

CleanupSend();

}

async Task SendClose(bool async, CancellationToken cancellationToken)

{

BeginSend();

var connector = Connection.Connector;

Debug.Assert(connector != null);

var buf = connector.WriteBuffer;

foreach (var statement in _statements.Where(s => s.IsPrepared))

{

if (FlushOccurred)

{

async = true;

SynchronizationContext.SetSynchronizationContext(SingleThreadSynchronizationContext);

}

await connector.CloseMessage

.Populate(StatementOrPortal.Statement, statement.StatementName)

.Write(buf, async);

Debug.Assert(statement.PreparedStatement != null);

statement.PreparedStatement.State = PreparedState.BeingUnprepared;

}

await SyncMessage.Instance.Write(buf, async);

await buf.Flush(async);

CleanupSend();

}

#endregion

#region Execute Non Query

/// <summary>

/// Executes a SQL statement against the connection and returns the number of rows affected.

/// </summary>

/// <returns>The number of rows affected if known; -1 otherwise.</returns>

public override int ExecuteNonQuery() => ExecuteNonQuery(false, CancellationToken.None).GetAwaiter().GetResult();

/// <summary>

/// Asynchronous version of <see cref="ExecuteNonQuery()"/>

/// </summary>

/// <param name="cancellationToken">The token to monitor for cancellation requests.</param>

/// <returns>A task representing the asynchronous operation, with the number of rows affected if known; -1 otherwise.</returns>

public override Task<int> ExecuteNonQueryAsync(CancellationToken cancellationToken)

=> SynchronizationContextSwitcher.NoContext(async () =>

{

cancellationToken.ThrowIfCancellationRequested();

using (cancellationToken.Register(Cancel))

return await ExecuteNonQuery(true, cancellationToken);

});

[MethodImpl(MethodImplOptions.AggressiveInlining)]

async Task<int> ExecuteNonQuery(bool async, CancellationToken cancellationToken)

{

var connector = CheckReadyAndGetConnector();

using (connector.StartUserAction(this))

using (var reader = await Execute(CommandBehavior.Default, async, cancellationToken))

{

while (async ? await reader.NextResultAsync(cancellationToken) : reader.NextResult()) {}

reader.Close();

return reader.RecordsAffected;

}

}

#endregion Execute Non Query

#region Execute Scalar

/// <summary>

/// Executes the query, and returns the first column of the first row

/// in the result set returned by the query. Extra columns or rows are ignored.

/// </summary>

/// <returns>The first column of the first row in the result set,

/// or a null reference if the result set is empty.</returns>

[CanBeNull]

public override object ExecuteScalar() => ExecuteScalar(false, CancellationToken.None).GetAwaiter().GetResult();

/// <summary>

/// Asynchronous version of <see cref="ExecuteScalar()"/>

/// </summary>

/// <param name="cancellationToken">The token to monitor for cancellation requests.</param>

/// <returns>A task representing the asynchronous operation, with the first column of the

/// first row in the result set, or a null reference if the result set is empty.</returns>

[ItemCanBeNull]

public override Task<object> ExecuteScalarAsync(CancellationToken cancellationToken)

=> SynchronizationContextSwitcher.NoContext(async () =>

{

cancellationToken.ThrowIfCancellationRequested() ;

using (cancellationToken.Register(Cancel))

return await ExecuteScalar(true, cancellationToken);

});

[MethodImpl(MethodImplOptions.AggressiveInlining)]

[ItemCanBeNull]

async ValueTask<object> ExecuteScalar(bool async, CancellationToken cancellationToken)

{

var connector = CheckReadyAndGetConnector();

var behavior = CommandBehavior.SingleRow;

if (!Parameters.HasOutputParameters)

behavior |= CommandBehavior.SequentialAccess;

using (connector.StartUserAction(this))

using (var reader = await Execute(behavior, async, cancellationToken))

return reader.Read() && reader.FieldCount != 0 ? reader.GetValue(0) : null;

}

#endregion Execute Scalar

#region Execute Reader

/// <summary>

/// Executes the CommandText against the Connection, and returns an DbDataReader.

/// </summary>

/// <remarks>

/// Unlike the ADO.NET method which it replaces, this method returns a Npgsql-specific

/// DataReader.

/// </remarks>

/// <returns>A DbDataReader object.</returns>

public new NpgsqlDataReader ExecuteReader() => (NpgsqlDataReader) base.ExecuteReader();

/// <summary>