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
Prev Previous commit
Next Next commit
Code size tweaks
  • Loading branch information
@NinoFloris
NinoFloris committed Mar 25, 2024
commit 845f8229d4b2e997773730c3a5a6a55bb9d80e4c
8 changes: 7 additions & 1 deletion src/Npgsql/Internal/NpgsqlReadBuffer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -417,7 +417,7 @@ internal NpgsqlReadBuffer AllocateOversize(int count)
/// <summary>
/// Skip a given number of bytes.
/// </summary>
internal void Skip(int len, bool allowIO = false)
internal void Skip(int len, bool allowIO)
{
Debug.Assert(len >= 0);

Expand All @@ -438,6 +438,12 @@ internal void Skip(int len, bool allowIO = false)
ReadPosition += len;
}

internal void Skip(int len)
{
Debug.Assert(ReadBytesLeft >= len);
ReadPosition += len;
}

/// <summary>
/// Skip a given number of bytes.
/// </summary>
Expand Down
61 changes: 27 additions & 34 deletions src/Npgsql/NpgsqlDataReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1426,8 +1426,7 @@ public override long GetBytes(int ordinal, long dataOffset, byte[]? buffer, int
if (buffer != null && (length < 0 || length > buffer.Length - bufferOffset))
ThrowHelper.ThrowIndexOutOfRangeException("bufferOffset must be between 0 and {0}", buffer.Length - bufferOffset);

var columnLength = SeekToColumn(ordinal, field.DataFormat, resumableOp: true);
if (columnLength == -1)
if (SeekToColumn(ordinal, field.DataFormat, resumableOp: true) is var columnLength && columnLength is -1)
ThrowHelper.ThrowInvalidCastException_NoValue(field);

if (buffer is null)
Expand Down Expand Up @@ -1915,9 +1914,9 @@ int SeekToColumn(int ordinal, DataFormat dataFormat, bool resumableOp = false)
}

reader.Commit();
var columnLength = BufferSeekToColumn(column, ordinal);
var columnLength = BufferSeekToColumn(column, ordinal, !_isRowBuffered);
reader.Init(columnLength, dataFormat, resumableOp);
return reader.FieldSize;
return columnLength;

static void ThrowInvalidSequentialSeek(int column, int ordinal)
=> ThrowHelper.ThrowInvalidOperationException(
Expand All @@ -1940,29 +1939,26 @@ async ValueTask<int> Core(int ordinal, DataFormat dataFormat, bool resumableOp)

var reader = PgReader;
await reader.CommitAsync().ConfigureAwait(false);
var columnLength = await BufferSeekToColumnAsync(_column, ordinal).ConfigureAwait(false);
var columnLength = await BufferSeekToColumnAsync(_column, ordinal, !_isRowBuffered).ConfigureAwait(false);
reader.Init(columnLength, dataFormat, resumableOp);
return columnLength;
}
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
int BufferSeekToColumn(int column, int ordinal)
int BufferSeekToColumn(int column, int ordinal, bool allowIO)
{
Debug.Assert(column < ordinal || _isRowBuffered);

var buffer = Buffer;
var columnLength = 0;
Debug.Assert(column < ordinal || !allowIO);

if (column >= ordinal)
{
columnLength = SeekBackwards(buffer, _columns, _columnsStartPos, ordinal);
_column = ordinal;
return columnLength;
return SeekBackwards(ordinal);
}

// We know we need at least one iteration, a do while also helps with optimal codegen.
var allowIO = !_isRowBuffered;
var buffer = Buffer;
var columnLength = 0;
do
{
if (columnLength > 0)
Expand All @@ -1972,38 +1968,36 @@ int BufferSeekToColumn(int column, int ordinal)
buffer.Ensure(sizeof(int));
columnLength = buffer.ReadInt32();
Debug.Assert(columnLength >= -1);
_column++;
} while (_column < ordinal);
} while (++_column < ordinal);

return columnLength;

// On the first call to SeekBackwards we'll fill up the columns list as we may need seek positions more than once.
[MethodImpl(MethodImplOptions.NoInlining)]
static int SeekBackwards(NpgsqlReadBuffer buffer, List<(int Offset, int Length)> columns, int columnsStartPos, int ordinal)
int SeekBackwards(int ordinal)
{
// Backfill the first column.
if (columns.Count is 0)
{
buffer.ReadPosition = columnsStartPos;
var len = buffer.ReadInt32();
columns.Add((buffer.ReadPosition, len));
}
for (var lastColumnRead = columns.Count; ordinal >= lastColumnRead; lastColumnRead++)
var buffer = Buffer;
var columns = _columns;

(buffer.ReadPosition, var columnLength) = columns.Count is 0
? (_columnsStartPos, 0)
: columns[Math.Min(columns.Count -1, ordinal)];

while (columns.Count <= ordinal)
{
(buffer.ReadPosition, var lastLen) = columns[lastColumnRead - 1];
if (lastLen > 0)
buffer.Skip(lastLen);
var len = buffer.ReadInt32();
columns.Add((buffer.ReadPosition, len));
if (columnLength > 0)
buffer.Skip(columnLength);
columnLength = buffer.ReadInt32();
columns.Add((buffer.ReadPosition, columnLength));
}
(buffer.ReadPosition, var columnLength) = columns[ordinal];

return columnLength;
}
}

ValueTask<int> BufferSeekToColumnAsync(int column, int ordinal)
ValueTask<int> BufferSeekToColumnAsync(int column, int ordinal, bool allowIO)
{
return column >= ordinal ? new(BufferSeekToColumn(column, ordinal)) : Core(ordinal);
return !allowIO || column >= ordinal ? new(BufferSeekToColumn(column, ordinal, allowIO)) : Core(ordinal);

[AsyncMethodBuilder(typeof(PoolingAsyncValueTaskMethodBuilder<>))]
async ValueTask<int> Core(int ordinal)
Expand All @@ -2019,8 +2013,7 @@ async ValueTask<int> Core(int ordinal)
await buffer.EnsureAsync(sizeof(int)).ConfigureAwait(false);
columnLength = buffer.ReadInt32();
Debug.Assert(columnLength >= -1);
_column++;
} while (_column < ordinal);
} while (++_column < ordinal);

return columnLength;
}
Expand Down