Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Add COPY operations dispose on initialization failure #6220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vonzshik merged 6 commits into from
Oct 2, 2025
Merged

Add COPY operations dispose on initialization failure #6220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9f789e6
Add COPY operations dispose on initialization failure
vonzshik Sep 23, 2025
4782d40
Fixes
vonzshik Sep 23, 2025
b640f04
A few missed calls
vonzshik Sep 23, 2025
24efaee
Review
vonzshik Oct 1, 2025
7084240
Fix state check
vonzshik Oct 1, 2025
19e7610
Remove superfluous comments
vonzshik Oct 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 21 additions & 8 deletions 29 src/Npgsql/NpgsqlBinaryExporter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ public sealed class NpgsqlBinaryExporter : ICancelable

NpgsqlConnector _connector;
NpgsqlReadBuffer _buf;
bool _isConsumed, _isDisposed;
ExporterState _state = ExporterState.Uninitialized;
long _endOfMessagePos;

short _column;
Expand Down Expand Up @@ -91,6 +91,7 @@ internal async Task Init(string copyToCommand, bool async, CancellationToken can
throw _connector.UnexpectedMessageReceived(msg.Code);
}

_state = ExporterState.Ready;
NumColumns = copyOutResponse.NumColumns;
_columnInfoCache = new PgConverterInfo[NumColumns];
_rowsExported = 0;
Expand Down Expand Up @@ -141,7 +142,7 @@ async Task ReadHeader(bool async)
async ValueTask<int> StartRow(bool async, CancellationToken cancellationToken = default)
{
ThrowIfDisposed();
if (_isConsumed)
if (_state == ExporterState.Consumed)
return -1;

using var registration = _connector.StartNestedCancellableOperation(cancellationToken);
Expand Down Expand Up @@ -176,7 +177,7 @@ async ValueTask<int> StartRow(bool async, CancellationToken cancellationToken =
Expect<CommandCompleteMessage>(await _connector.ReadMessage(async).ConfigureAwait(false), _connector);
Expect<ReadyForQueryMessage>(await _connector.ReadMessage(async).ConfigureAwait(false), _connector);
_column = BeforeRow;
_isConsumed = true;
_state = ExporterState.Consumed;
return -1;
}

Expand Down Expand Up @@ -437,7 +438,7 @@ void ThrowIfNotOnRow()

void ThrowIfDisposed()
{
if (_isDisposed)
if (_state == ExporterState.Disposed)
ThrowHelper.ThrowObjectDisposedException(nameof(NpgsqlBinaryExporter), "The COPY operation has already ended.");
}

Expand Down Expand Up @@ -472,10 +473,10 @@ public Task CancelAsync()

async ValueTask DisposeAsync(bool async)
{
if (_isDisposed)
if (_state == ExporterState.Disposed)
return;

if (_isConsumed)
if (_state is ExporterState.Consumed or ExporterState.Uninitialized)
{
LogMessages.BinaryCopyOperationCompleted(_copyLogger, _rowsExported, _connector.Id);
}
Expand Down Expand Up @@ -512,7 +513,7 @@ async ValueTask DisposeAsync(bool async)

void Cleanup()
{
Debug.Assert(!_isDisposed);
Debug.Assert(_state != ExporterState.Disposed);
var connector = _connector;

if (!ReferenceEquals(connector, null))
Expand All @@ -523,9 +524,21 @@ void Cleanup()
}

_buf = null!;
_isDisposed = true;
_state = ExporterState.Disposed;
}
}

#endregion

#region Enums

enum ExporterState
{
Uninitialized,
Ready,
Consumed,
Disposed
}

#endregion Enums
}
6 changes: 5 additions & 1 deletion 6 src/Npgsql/NpgsqlBinaryImporter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ public sealed class NpgsqlBinaryImporter : ICancelable
NpgsqlConnector _connector;
NpgsqlWriteBuffer _buf;

ImporterState _state;
ImporterState _state = ImporterState.Uninitialized;

/// <summary>
/// The number of columns in the current (not-yet-written) row.
Expand Down Expand Up @@ -99,6 +99,7 @@ internal async Task Init(string copyFromCommand, bool async, CancellationToken c
throw _connector.UnexpectedMessageReceived(msg.Code);
}

_state = ImporterState.Ready;
_params = new NpgsqlParameter[copyInResponse.NumColumns];
_rowsImported = 0;
_buf.StartCopyMode();
Expand Down Expand Up @@ -512,6 +513,7 @@ async ValueTask CloseAsync(bool async, CancellationToken cancellationToken = def
case ImporterState.Ready:
await Cancel(async, cancellationToken).ConfigureAwait(false);
break;
case ImporterState.Uninitialized:
case ImporterState.Cancelled:
case ImporterState.Committed:
break;
Expand Down Expand Up @@ -553,6 +555,7 @@ void CheckReady()
static void Throw(ImporterState state)
=> throw (state switch
{
ImporterState.Uninitialized => throw new InvalidOperationException("The COPY operation has not been initialized."),
ImporterState.Disposed => new ObjectDisposedException(typeof(NpgsqlBinaryImporter).FullName,
"The COPY operation has already ended."),
ImporterState.Cancelled => new InvalidOperationException("The COPY operation has already been cancelled."),
Expand All @@ -567,6 +570,7 @@ static void Throw(ImporterState state)

enum ImporterState
{
Uninitialized,
Ready,
Committed,
Cancelled,
Expand Down
75 changes: 60 additions & 15 deletions 75 src/Npgsql/NpgsqlConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1160,17 +1160,26 @@ async Task<NpgsqlBinaryImporter> BeginBinaryImport(bool async, string copyFromCo
LogMessages.StartingBinaryImport(connector.LoggingConfiguration.CopyLogger, connector.Id);
// no point in passing a cancellationToken here, as we register the cancellation in the Init method
connector.StartUserAction(ConnectorState.Copy, attemptPgCancellation: false);
var importer = new NpgsqlBinaryImporter(connector);
try
{
var importer = new NpgsqlBinaryImporter(connector);
await importer.Init(copyFromCommand, async, cancellationToken).ConfigureAwait(false);
connector.CurrentCopyOperation = importer;
return importer;
}
catch
{
connector.EndUserAction();
EndBindingScope(ConnectorBindingScope.Copy);
try
{
if (async)
await importer.DisposeAsync().ConfigureAwait(false);
else
importer.Dispose();
}
catch
{
// ignored
}
throw;
}
}
Expand Down Expand Up @@ -1210,17 +1219,26 @@ async Task<NpgsqlBinaryExporter> BeginBinaryExport(bool async, string copyToComm
LogMessages.StartingBinaryExport(connector.LoggingConfiguration.CopyLogger, connector.Id);
// no point in passing a cancellationToken here, as we register the cancellation in the Init method
connector.StartUserAction(ConnectorState.Copy, attemptPgCancellation: false);
var exporter = new NpgsqlBinaryExporter(connector);
try
{
var exporter = new NpgsqlBinaryExporter(connector);
await exporter.Init(copyToCommand, async, cancellationToken).ConfigureAwait(false);
connector.CurrentCopyOperation = exporter;
return exporter;
}
catch
{
connector.EndUserAction();
EndBindingScope(ConnectorBindingScope.Copy);
try
{
if (async)
await exporter.DisposeAsync().ConfigureAwait(false);
else
exporter.Dispose();
}
catch
{
// ignored
}
throw;
}
}
Expand Down Expand Up @@ -1266,18 +1284,27 @@ async Task<TextWriter> BeginTextImport(bool async, string copyFromCommand, Cance
LogMessages.StartingTextImport(connector.LoggingConfiguration.CopyLogger, connector.Id);
// no point in passing a cancellationToken here, as we register the cancellation in the Init method
connector.StartUserAction(ConnectorState.Copy, attemptPgCancellation: false);
var copyStream = new NpgsqlRawCopyStream(connector);
try
{
var copyStream = new NpgsqlRawCopyStream(connector);
await copyStream.Init(copyFromCommand, async, cancellationToken).ConfigureAwait(false);
var writer = new NpgsqlCopyTextWriter(connector, copyStream);
connector.CurrentCopyOperation = writer;
return writer;
}
catch
{
connector.EndUserAction();
EndBindingScope(ConnectorBindingScope.Copy);
try
{
if (async)
await copyStream.DisposeAsync().ConfigureAwait(false);
else
copyStream.Dispose();
}
catch
{
// ignored
}
throw;
}
}
Expand Down Expand Up @@ -1323,18 +1350,27 @@ async Task<TextReader> BeginTextExport(bool async, string copyToCommand, Cancell
LogMessages.StartingTextExport(connector.LoggingConfiguration.CopyLogger, connector.Id);
// no point in passing a cancellationToken here, as we register the cancellation in the Init method
connector.StartUserAction(ConnectorState.Copy, attemptPgCancellation: false);
var copyStream = new NpgsqlRawCopyStream(connector);
try
{
var copyStream = new NpgsqlRawCopyStream(connector);
await copyStream.Init(copyToCommand, async, cancellationToken).ConfigureAwait(false);
var reader = new NpgsqlCopyTextReader(connector, copyStream);
connector.CurrentCopyOperation = reader;
return reader;
}
catch
{
connector.EndUserAction();
EndBindingScope(ConnectorBindingScope.Copy);
try
{
if (async)
await copyStream.DisposeAsync().ConfigureAwait(false);
else
copyStream.Dispose();
}
catch
{
// ignored
}
throw;
}
}
Expand Down Expand Up @@ -1380,9 +1416,9 @@ async Task<NpgsqlRawCopyStream> BeginRawBinaryCopy(bool async, string copyComman
LogMessages.StartingRawCopy(connector.LoggingConfiguration.CopyLogger, connector.Id);
// no point in passing a cancellationToken here, as we register the cancellation in the Init method
connector.StartUserAction(ConnectorState.Copy, attemptPgCancellation: false);
var stream = new NpgsqlRawCopyStream(connector);
try
{
var stream = new NpgsqlRawCopyStream(connector);
await stream.Init(copyCommand, async, cancellationToken).ConfigureAwait(false);
if (!stream.IsBinary)
{
Expand All @@ -1395,8 +1431,17 @@ async Task<NpgsqlRawCopyStream> BeginRawBinaryCopy(bool async, string copyComman
}
catch
{
connector.EndUserAction();
EndBindingScope(ConnectorBindingScope.Copy);
try
{
if (async)
await stream.DisposeAsync().ConfigureAwait(false);
else
stream.Dispose();
}
catch
{
// ignored
}
throw;
}
}
Expand Down
Loading
Morty Proxy This is a proxified and sanitized view of the page, visit original site.