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

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
Pull reader into a local
  • Loading branch information
@NinoFloris
NinoFloris committed Mar 26, 2024
commit cf122e5d6480ec628a285c660283ae389f864717
47 changes: 26 additions & 21 deletions src/Npgsql/NpgsqlDataReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1441,15 +1441,16 @@ public override long GetBytes(int ordinal, long dataOffset, byte[]? buffer, int
if (buffer is null)
return columnLength;

var reader = PgReader;
// Move to offset
if (_isSequential && PgReader.FieldOffset > dataOffset)
if (_isSequential && reader.FieldOffset > dataOffset)
ThrowHelper.ThrowInvalidOperationException("Attempt to read a position in the column which has already been read");

PgReader.Seek((int)dataOffset);
reader.Seek((int)dataOffset);

// At offset, read into buffer.
length = Math.Min(length, PgReader.FieldRemaining);
PgReader.ReadBytes(new Span<byte>(buffer, bufferOffset, length));
length = Math.Min(length, reader.FieldRemaining);
reader.ReadBytes(new Span<byte>(buffer, bufferOffset, length));
return length;
}

Expand Down Expand Up @@ -1503,20 +1504,21 @@ public override long GetChars(int ordinal, long dataOffset, char[]? buffer, int
if (SeekToColumn(ordinal, format, resumableOp: true) is -1)
ThrowHelper.ThrowInvalidCastException_NoValue(RowDescription[ordinal]);

var reader = PgReader;
dataOffset = buffer is null ? 0 : dataOffset;
PgReader.InitCharsRead(checked((int)dataOffset),
reader.InitCharsRead(checked((int)dataOffset),
buffer is not null ? new ArraySegment<char>(buffer, bufferOffset, length) : (ArraySegment<char>?)null,
out var previousDataOffset);

if (_isSequential && previousDataOffset > dataOffset)
ThrowHelper.ThrowInvalidOperationException("Attempt to read a position in the column which has already been read");

PgReader.StartRead(bufferRequirement);
reader.StartRead(bufferRequirement);
var result = asObject
? (GetChars)converter.ReadAsObject(PgReader)
: ((PgConverter<GetChars>)converter).Read(PgReader);
PgReader.AdvanceCharsRead(result.Read);
PgReader.EndRead();
? (GetChars)converter.ReadAsObject(reader)
: ((PgConverter<GetChars>)converter).Read(reader);
reader.AdvanceCharsRead(result.Read);
reader.EndRead();
return result.Read;
}

Expand Down Expand Up @@ -1576,11 +1578,12 @@ async Task<T> Core(int ordinal, CancellationToken cancellationToken)
if (await SeekToColumnAsync(ordinal, field).ConfigureAwait(false) is -1)
return DbNullValueOrThrow<T>(ordinal);

await PgReader.StartReadAsync(bufferRequirement, cancellationToken).ConfigureAwait(false);
var reader = PgReader;
await reader.StartReadAsync(bufferRequirement, cancellationToken).ConfigureAwait(false);
var result = asObject
? (T)await converter.ReadAsObjectAsync(PgReader, cancellationToken).ConfigureAwait(false)
: await converter.UnsafeDowncast<T>().ReadAsync(PgReader, cancellationToken).ConfigureAwait(false);
await PgReader.EndReadAsync().ConfigureAwait(false);
? (T)await converter.ReadAsObjectAsync(reader, cancellationToken).ConfigureAwait(false)
: await converter.UnsafeDowncast<T>().ReadAsync(reader, cancellationToken).ConfigureAwait(false);
await reader.EndReadAsync().ConfigureAwait(false);
return result;
}
}
Expand All @@ -1601,11 +1604,12 @@ T GetFieldValueCore<T>(int ordinal)
if (SeekToColumn(ordinal, field) is -1)
return DbNullValueOrThrow<T>(ordinal);

PgReader.StartRead(bufferRequirement);
var reader = PgReader;
reader.StartRead(bufferRequirement);
var result = asObject
? (T)converter.ReadAsObject(PgReader)
: converter.UnsafeDowncast<T>().Read(PgReader);
PgReader.EndRead();
? (T)converter.ReadAsObject(reader)
: converter.UnsafeDowncast<T>().Read(reader);
reader.EndRead();
return result;
}

Expand All @@ -1625,9 +1629,10 @@ public override object GetValue(int ordinal)
if (SeekToColumn(ordinal, field) is -1)
return DBNull.Value;

PgReader.StartRead(bufferRequirement);
var result = converter.ReadAsObject(PgReader);
PgReader.EndRead();
var reader = PgReader;
reader.StartRead(bufferRequirement);
var result = converter.ReadAsObject(reader);
reader.EndRead();

return result;
}
Expand Down