Skip to content

Rework row reading #5631

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

Draft
NinoFloris wants to merge 12 commits into
base: main
Choose a base branch
from
Draft

Rework row reading #5631

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
d72b934
Move Skip async arg to the start
NinoFloris Dec 5, 2023
e9dcab9
Separate resuming and rewinding from committing proper
NinoFloris Dec 5, 2023
d8c0efa
Refactor SeekToColumn to a sync and an async variant
NinoFloris Dec 5, 2023
436e6b0
Improve bounds checks
NinoFloris Dec 5, 2023
4a65463
Perf
NinoFloris Dec 6, 2023
57d8264
Make sure we don't consume data in the stream during reader restart
NinoFloris Dec 7, 2023
73de110
Split buffer and reader logic entirely
NinoFloris Dec 7, 2023
845f822
Code size tweaks
NinoFloris Mar 25, 2024
3e8d463
Simplify by resolving Stream by default as a converter
NinoFloris Dec 3, 2023
cf122e5
Pull reader into a local
NinoFloris Dec 8, 2023
fe448b4
Rework row reading
NinoFloris Dec 7, 2023
191c0cf
Include column count into parsed message
NinoFloris Mar 26, 2024
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
Next Next commit
Move Skip async arg to the start
  • Loading branch information
@NinoFloris
NinoFloris committed Mar 25, 2024
commit d72b9347fb635d98013dfb1e96bb8d5975b44d95
2 changes: 1 addition & 1 deletion src/Npgsql/Internal/NpgsqlConnector.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1294,7 +1294,7 @@ internal ValueTask<IBackendMessage> ReadMessage(
{
if (dataRowLoadingMode == DataRowLoadingMode.Skip)
{
await ReadBuffer.Skip(len, async).ConfigureAwait(false);
await ReadBuffer.Skip(async, len).ConfigureAwait(false);
continue;
}
}
Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/Internal/NpgsqlReadBuffer.Stream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,7 +210,7 @@ async ValueTask DisposeAsync(bool disposing, bool async)
var pos = _buf.CumulativeReadPosition - _startPos;
var remaining = checked((int)(CurrentLength - pos));
if (remaining > 0)
await _buf.Skip(remaining, async).ConfigureAwait(false);
await _buf.Skip(async, remaining).ConfigureAwait(false);
}

IsDisposed = true;
Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/Internal/NpgsqlReadBuffer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,7 +426,7 @@ internal void Skip(int len)
/// <summary>
/// Skip a given number of bytes.
/// </summary>
public async Task Skip(int len, bool async)
public async Task Skip(bool async, int len)
{
Debug.Assert(len >= 0);

Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/Internal/PgReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -552,7 +552,7 @@ internal async ValueTask Consume(bool async, int? count = null, CancellationToke
var origOffset = FieldOffset;
// A breaking exception unwind from a nested scope should not try to consume its remaining data.
if (!_buffer.Connector.IsBroken)
await _buffer.Skip(remaining, async).ConfigureAwait(false);
await _buffer.Skip(async, remaining).ConfigureAwait(false);

Debug.Assert(FieldRemaining == FieldSize - origOffset - remaining);
}
Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/NpgsqlBinaryExporter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -492,7 +492,7 @@ async ValueTask DisposeAsync(bool async)
else
PgReader.Commit(resuming: false);
// Finish the current CopyData message
await _buf.Skip(checked((int)(_endOfMessagePos - _buf.CumulativeReadPosition)), async).ConfigureAwait(false);
await _buf.Skip(async, checked((int)(_endOfMessagePos - _buf.CumulativeReadPosition))).ConfigureAwait(false);
// Read to the end
_connector.SkipUntil(BackendMessageCode.CopyDone);
// We intentionally do not pass a CancellationToken since we don't want to cancel cleanup
Expand Down
4 changes: 2 additions & 2 deletions src/Npgsql/NpgsqlDataReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2065,7 +2065,7 @@ async ValueTask<int> Core(bool async, bool reread, bool commit, int ordinal, Dat
_column++;
Debug.Assert(columnLength >= -1);
if (columnLength > 0)
await buffer.Skip(columnLength, async).ConfigureAwait(false);
await buffer.Skip(async: true, columnLength).ConfigureAwait(false);
}

await buffer.Ensure(4, async).ConfigureAwait(false);
Expand Down Expand Up @@ -2144,7 +2144,7 @@ async Task ConsumeRowSequential(bool async)
_column++;
Debug.Assert(columnLength >= -1);
if (columnLength > 0)
await buffer.Skip(columnLength, async).ConfigureAwait(false);
await buffer.Skip(async, columnLength).ConfigureAwait(false);
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/NpgsqlRawCopyStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,7 +379,7 @@ async ValueTask DisposeAsync(bool disposing, bool async)
{
if (_leftToReadInDataMsg > 0)
{
await _readBuf.Skip(_leftToReadInDataMsg, async).ConfigureAwait(false);
await _readBuf.Skip(async, _leftToReadInDataMsg).ConfigureAwait(false);
}
_connector.SkipUntil(BackendMessageCode.ReadyForQuery);
}
Expand Down
2 changes: 1 addition & 1 deletion src/Npgsql/Replication/ReplicationConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -499,7 +499,7 @@ internal async IAsyncEnumerator<XLogDataMessage> StartReplicationInternal(
// Our consumer may not have read the stream to the end, but it might as well have been us
// ourselves bypassing the stream and reading directly from the buffer in StartReplication()
if (!columnStream.IsDisposed && columnStream.Position < columnStream.Length && !bypassingStream)
await buf.Skip(checked((int)(columnStream.Length - columnStream.Position)), true).ConfigureAwait(false);
await buf.Skip(async: true, checked((int)(columnStream.Length - columnStream.Position))).ConfigureAwait(false);

continue;
}
Expand Down