From 351eaad48f3db8f260fb5d567bd3ccd9c8153094 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 12:45:27 +0000 Subject: [PATCH 1/4] Initial plan From 5a22cbf1508196e9e6ef68667240725b8bc9c021 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 12:50:10 +0000 Subject: [PATCH 2/4] fix: restrict BSON scans to collection-owned locations Co-authored-by: mrdevrobot <12503462+mrdevrobot@users.noreply.github.com> --- .../Collections/DocumentCollection.cs | 269 ++++++++++-------- .../CrossCollectionQueryIsolationTests.cs | 72 +++++ 2 files changed, 219 insertions(+), 122 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 7bb3e16a..e6cccbf1 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -351,39 +351,121 @@ private async Task ApplyRetentionPolicyCoreAsync(CancellationToken ct) /// /// Reads the raw BSON bytes for a document at the given location without deserializing. - /// Returns null if the slot is deleted, the page is invalid, or the document spans - /// multiple overflow pages — overflow documents are exempt from age-based retention - /// because reassembling the full payload is expensive in this context. - /// For retention purposes, and - /// still apply to overflow documents (their primary-index entry is included in the count scan). + /// Returns null if the slot is deleted or the page/location is invalid. + /// Overflow documents are reassembled into a single BSON payload when needed. /// - private byte[]? ReadRawBytesAt(DocumentLocation location, ulong txnId) + private static bool TryReadInlineRawBytes(byte[] pageBuffer, ushort slotIndex, out ReadOnlySpan rawBytes) { - var buffer = ArrayPool.Shared.Rent(_storage.PageSize); + rawBytes = default; + + if ((PageType)pageBuffer[4] != PageType.Data) + return false; + + var header = SlottedPageHeader.ReadFrom(pageBuffer); + if (slotIndex >= header.SlotCount) return false; + + var slotOffset = SlottedPageHeader.Size + (slotIndex * SlotEntry.Size); + var slot = SlotEntry.ReadFrom(pageBuffer.AsSpan(slotOffset)); + if ((slot.Flags & (SlotFlags.Deleted | SlotFlags.HasOverflow)) != 0) return false; + if (slot.Offset + slot.Length > pageBuffer.Length) return false; + + rawBytes = pageBuffer.AsSpan(slot.Offset, slot.Length); + return true; + } + + private byte[]? ReadRawBytesAt(DocumentLocation location, ulong txnId, byte[]? preloadedPage = null) + { + byte[]? ownedBuffer = null; + var buffer = preloadedPage ?? (ownedBuffer = ArrayPool.Shared.Rent(_storage.PageSize)); try { - _storage.ReadPage(location.PageId, txnId, buffer); + if (ownedBuffer != null) + _storage.ReadPage(location.PageId, txnId, buffer); + var pageType = (PageType)buffer[4]; if (pageType == PageType.Free || pageType == PageType.Empty || pageType == PageType.TimeSeries) return null; + if (TryReadInlineRawBytes(buffer, location.SlotIndex, out var rawBytes)) + return rawBytes.ToArray(); + var header = SlottedPageHeader.ReadFrom(buffer); if (location.SlotIndex >= header.SlotCount) return null; var slotOffset = SlottedPageHeader.Size + (location.SlotIndex * SlotEntry.Size); var slot = SlotEntry.ReadFrom(buffer.AsSpan(slotOffset)); if ((slot.Flags & SlotFlags.Deleted) != 0) return null; - if ((slot.Flags & SlotFlags.HasOverflow) != 0) return null; // skip overflow docs for retention + if ((slot.Flags & SlotFlags.HasOverflow) == 0) return null; + + if (slot.Offset + slot.Length > buffer.Length || slot.Length < 8) return null; + + var payload = buffer.AsSpan(slot.Offset, slot.Length); + int totalLength = System.Buffers.Binary.BinaryPrimitives.ReadInt32LittleEndian(payload.Slice(0, 4)); + uint nextOverflowPageId = System.Buffers.Binary.BinaryPrimitives.ReadUInt32LittleEndian(payload.Slice(4, 4)); + if (totalLength <= 0) return null; + + int primaryChunkSize = slot.Length - 8; + if (primaryChunkSize < 0 || primaryChunkSize > totalLength) return null; + + var fullBuffer = new byte[totalLength]; + payload.Slice(8, primaryChunkSize).CopyTo(fullBuffer); + + int offset = primaryChunkSize; + var overflowBuffer = ArrayPool.Shared.Rent(_storage.PageSize); + try + { + var currentOverflowPageId = nextOverflowPageId; + while (currentOverflowPageId != 0 && offset < totalLength) + { + _storage.ReadPage(currentOverflowPageId, txnId, overflowBuffer); + var overflowHeader = SlottedPageHeader.ReadFrom(overflowBuffer); - if (slot.Offset + slot.Length > buffer.Length) return null; - return buffer.AsSpan(slot.Offset, slot.Length).ToArray(); + int chunkSize = Math.Min(_storage.PageSize - SlottedPageHeader.Size, totalLength - offset); + overflowBuffer.AsSpan(SlottedPageHeader.Size, chunkSize) + .CopyTo(fullBuffer.AsSpan(offset)); + + offset += chunkSize; + currentOverflowPageId = overflowHeader.NextOverflowPage; + } + } + finally + { + ArrayPool.Shared.Return(overflowBuffer); + } + + return offset == totalLength ? fullBuffer : null; } finally { - ArrayPool.Shared.Return(buffer); + if (ownedBuffer != null) + ArrayPool.Shared.Return(ownedBuffer); } } + private async Task>> GetCollectionLocationsByPageAsync( + ulong txnId, + CancellationToken ct = default) + { + var locationsByPage = new Dictionary>(); + + await foreach (var entry in _primaryIndex + .RangeAsync(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId, ct) + .ConfigureAwait(false)) + { + ct.ThrowIfCancellationRequested(); + + if (!locationsByPage.TryGetValue(entry.Location.PageId, out var locations)) + { + locations = new List(); + locationsByPage[entry.Location.PageId] = locations; + } + + locations.Add(entry.Location); + } + + return locationsByPage; + } + /// /// Extracts a UTC-ticks timestamp from raw BSON bytes using the given field name. /// Resolves the field name against the Key Dictionary for correct C-BSON lookup, @@ -895,54 +977,30 @@ public async IAsyncEnumerable ScanAsync( var sw = _storage.MetricsDispatcher != null ? ValueStopwatch.StartNew() : default; var txnId = transaction?.TransactionId ?? 0UL; + var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); var buffer = ArrayPool.Shared.Rent(_storage.PageSize); + var keyMap = _storage.GetKeyReverseMap(); try { - foreach (var pageId in _storage.GetCollectionPageIds(_collectionName)) + foreach (var (pageId, locations) in locationsByPage) { ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct); - - var header = SlottedPageHeader.ReadFrom(buffer); - if (header.PageType != PageType.Data) continue; + await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - // Collect matching locations first (no Span across yield) - var matchingLocations = new List<(uint pageId, ushort slotIndex)>(); + foreach (var location in locations) { - var slots = MemoryMarshal.Cast( - buffer.AsSpan(SlottedPageHeader.Size, header.SlotCount * SlotEntry.Size)); - - for (int i = 0; i < header.SlotCount; i++) + if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) { - var slot = slots[i]; - if (slot.Flags.HasFlag(SlotFlags.Deleted)) continue; - - var data = buffer.AsSpan(slot.Offset, slot.Length); - var reader = new BsonSpanReader(data, _storage.GetKeyReverseMap()); - - if (predicate(reader)) - { - matchingLocations.Add((pageId, (ushort)i)); - } + var rawBytes = ReadRawBytesAt(location, txnId, buffer); + if (rawBytes is null) continue; + inlineRawBytes = rawBytes; } - } - // Yield matching documents. The buffer already holds this page — pass it - // to FindByLocationAsync so the page is not read a second time. - // A data page may contain slots from multiple collections; if the mapper - // fails for a foreign-collection document, silently skip it. - foreach (var (pid, idx) in matchingLocations) - { - T? doc; - try - { - doc = await FindByLocationAsync(new DocumentLocation(pid, idx), txnId, buffer, ct); - } - catch - { - continue; // foreign-collection document — skip - } + var reader = new BsonSpanReader(inlineRawBytes, keyMap); + if (!predicate(reader)) continue; + + var doc = await FindByLocationAsync(location, txnId, buffer, ct).ConfigureAwait(false); if (doc != null) yield return doc; } } @@ -1105,40 +1163,29 @@ public async IAsyncEnumerable ScanAsync( if (projector == null) throw new ArgumentNullException(nameof(projector)); var txnId = 0UL; + var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); var buffer = ArrayPool.Shared.Rent(_storage.PageSize); + var keyMap = _storage.GetKeyReverseMap(); try { - foreach (var pageId in _storage.GetCollectionPageIds(_collectionName)) + foreach (var (pageId, locations) in locationsByPage) { ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct); + await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - var header = SlottedPageHeader.ReadFrom(buffer); - if (header.PageType != PageType.Data) continue; - - // Process all slots and collect results (no Span across yield) - var pageResults = new List(); + foreach (var location in locations) { - var slots = MemoryMarshal.Cast( - buffer.AsSpan(SlottedPageHeader.Size, header.SlotCount * SlotEntry.Size)); - - var keyMap = _storage.GetKeyReverseMap(); - for (int i = 0; i < header.SlotCount; i++) + if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) { - var slot = slots[i]; - if (slot.Flags.HasFlag(SlotFlags.Deleted)) continue; - - var data = buffer.AsSpan(slot.Offset, slot.Length); - var reader = new BsonSpanReader(data, keyMap); - var result = projector(reader); - if (result is not null) pageResults.Add(result); + var rawBytes = ReadRawBytesAt(location, txnId, buffer); + if (rawBytes is null) continue; + inlineRawBytes = rawBytes; } - } - // Yield results after Span is out of scope - foreach (var result in pageResults) - { + var result = projector(new BsonSpanReader(inlineRawBytes, keyMap)); + if (result is null) continue; + yield return result; } } @@ -1154,11 +1201,9 @@ public async IAsyncEnumerable ScanAsync( /// directly on raw BSON bytes — no CLR instances are ever created. /// /// - /// The predicate is executed once per non-deleted, non-overflow slot on every data page. - /// Overflow-flagged primary slots are skipped: the BSON data in such slots starts after - /// an 8-byte overflow header that a normal BSON predicate cannot parse correctly, so - /// those documents are not counted (same behaviour as the existing - /// for overflow documents). + /// The predicate is executed once per document referenced by the collection's primary index. + /// Raw BSON is read directly from the owning slot, reassembling overflow documents when needed, + /// and no CLR instances are created. /// internal async Task CountScanAsync( BsonReaderPredicate predicate, @@ -1167,40 +1212,34 @@ internal async Task CountScanAsync( if (predicate == null) throw new ArgumentNullException(nameof(predicate)); var txnId = 0UL; + var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); var buffer = ArrayPool.Shared.Rent(_storage.PageSize); int count = 0; + var keyMap = _storage.GetKeyReverseMap(); try { - foreach (var pageId in _storage.GetCollectionPageIds(_collectionName)) + foreach (var (pageId, locations) in locationsByPage) { ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct); - - var header = SlottedPageHeader.ReadFrom(buffer); - if (header.PageType != PageType.Data) continue; + await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - var slots = MemoryMarshal.Cast( - buffer.AsSpan(SlottedPageHeader.Size, header.SlotCount * SlotEntry.Size)); - var keyMap = _storage.GetKeyReverseMap(); - - for (int i = 0; i < header.SlotCount; i++) + foreach (var location in locations) { - var slot = slots[i]; - if ((slot.Flags & SlotFlags.Deleted) != 0) continue; - // Skip overflow continuation slots: the primary slot's raw data starts - // with an 8-byte overflow header (totalLength + nextPageId), not BSON. - if ((slot.Flags & SlotFlags.HasOverflow) != 0) continue; + if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) + { + var rawBytes = ReadRawBytesAt(location, txnId, buffer); + if (rawBytes is null) continue; + inlineRawBytes = rawBytes; + } - var data = buffer.AsSpan(slot.Offset, slot.Length); - var reader = new BsonSpanReader(data, keyMap); try { - if (predicate(reader)) count++; + if (predicate(new BsonSpanReader(inlineRawBytes, keyMap))) count++; } catch { - // Malformed BSON or foreign-collection slot — skip silently. + // Malformed BSON — skip silently. } } } @@ -1259,7 +1298,8 @@ public async IAsyncEnumerable ParallelScanAsync( if (predicate == null) throw new ArgumentNullException(nameof(predicate)); var txnId = 0UL; - var allPageIds = _storage.GetCollectionPageIds(_collectionName).ToArray(); + var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); + var allPageIds = locationsByPage.Keys.ToArray(); var pageCount = allPageIds.Length; if (degreeOfParallelism <= 0) @@ -1267,6 +1307,7 @@ public async IAsyncEnumerable ParallelScanAsync( var semaphore = new SemaphoreSlim(degreeOfParallelism); var tasks = new List>>(); + var keyMap = _storage.GetKeyReverseMap(); for (int pageIdx = 0; pageIdx < pageCount; pageIdx++) { @@ -1284,34 +1325,19 @@ public async IAsyncEnumerable ParallelScanAsync( { await _storage.ReadPageAsync(localPageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct); - var header = SlottedPageHeader.ReadFrom(buffer); - if (header.PageType != PageType.Data) return results; - - // First pass: collect matching locations (Span scope) - var matchingIndices = new List(); + foreach (var location in locationsByPage[localPageId]) { - var slots = MemoryMarshal.Cast( - buffer.AsSpan(SlottedPageHeader.Size, header.SlotCount * SlotEntry.Size)); - - for (int i = 0; i < header.SlotCount; i++) + if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) { - var slot = slots[i]; - if (slot.Flags.HasFlag(SlotFlags.Deleted)) continue; - - var data = buffer.AsSpan(slot.Offset, slot.Length); - var reader = new BsonSpanReader(data, _storage.GetKeyReverseMap()); - - if (predicate(reader)) - { - matchingIndices.Add((ushort)i); - } + var rawBytes = ReadRawBytesAt(location, txnId, buffer); + if (rawBytes is null) continue; + inlineRawBytes = rawBytes; } - } - // Second pass: fetch documents (no Span, safe to await) - foreach (var idx in matchingIndices) - { - var doc = await FindByLocationAsync(new DocumentLocation(localPageId, idx), txnId, ct); + if (!predicate(new BsonSpanReader(inlineRawBytes, keyMap))) + continue; + + var doc = await FindByLocationAsync(location, txnId, buffer, ct).ConfigureAwait(false); if (doc != null) results.Add(doc); } } @@ -3826,4 +3852,3 @@ public ValueTask DeleteBulkAsync(IEnumerable ids, CancellationToken ct return DeleteBulkAsync(ids, null, ct); } } - diff --git a/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs b/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs index fcc5f016..1d952a7f 100644 --- a/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs +++ b/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs @@ -1,3 +1,4 @@ +using BLite.Core.Query; using BLite.Shared; namespace BLite.Tests; @@ -69,6 +70,77 @@ public async Task Contains_WithDuplicateValues_DoesNotDoubleCount() Assert.Equal(2, count); } + [Fact] + public async Task Where_OnUnindexedSharedField_DoesNotReturnCrossCollectionRows() + { + await _db.IntEntities.InsertAsync(new IntEntity { Id = 1, Name = "Pranzo" }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Pranzo", Age = 20 }); + await _db.SaveChangesAsync(); + + var results = await _db.IntEntities.AsQueryable() + .Where(x => x.Name == "Pranzo") + .ToListAsync(); + + Assert.Single(results); + Assert.Equal(1, results[0].Id); + Assert.Equal("Pranzo", results[0].Name); + } + + [Fact] + public async Task Count_OnUnindexedSharedField_DoesNotCountCrossCollectionRows() + { + await _db.IntEntities.InsertAsync(new IntEntity { Id = 1, Name = "Pranzo" }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Pranzo", Age = 20 }); + await _db.SaveChangesAsync(); + + var count = _db.IntEntities.AsQueryable() + .Count(x => x.Name == "Pranzo"); + + Assert.Equal(1, count); + } + + [Fact] + public async Task Max_OnUnindexedSharedField_DoesNotReadCrossCollectionRows() + { + await _db.Users.InsertAsync(new User { Name = "Alpha", Age = 30 }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Zulu", Age = 20 }); + await _db.SaveChangesAsync(); + + var maxName = _db.Users.AsQueryable().Max(x => x.Name); + + Assert.Equal("Alpha", maxName); + } + + [Fact] + public async Task ParallelScan_OnSharedPage_OnlyReturnsCurrentCollectionRows() + { + await _db.Users.InsertAsync(new User { Name = "Pranzo", Age = 30 }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Pranzo", Age = 20 }); + await _db.SaveChangesAsync(); + + var results = await _db.Users.ParallelScanAsync(reader => + { + reader.ReadDocumentSize(); + while (reader.Remaining > 0) + { + var type = reader.ReadBsonType(); + if (type == 0) break; + + var fieldName = reader.ReadElementHeader(); + if (fieldName == "name") + return reader.ReadString() == "Pranzo"; + + reader.SkipValue(type); + } + + return false; + }, degreeOfParallelism: 2).ToListAsync(); + + Assert.Single(results); + Assert.Equal("Pranzo", results[0].Name); + Assert.Equal(30, results[0].Age); + } + public void Dispose() { _db.Dispose(); From c65f7f17befa1e4b3e3c5dc9d35029675e8dfb9b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 12:54:02 +0000 Subject: [PATCH 3/4] perf: stream parallel scan batches from primary index Co-authored-by: mrdevrobot <12503462+mrdevrobot@users.noreply.github.com> --- .../Collections/DocumentCollection.cs | 46 +++++++++++++------ 1 file changed, 33 insertions(+), 13 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index e6cccbf1..532e24da 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -1298,35 +1298,42 @@ public async IAsyncEnumerable ParallelScanAsync( if (predicate == null) throw new ArgumentNullException(nameof(predicate)); var txnId = 0UL; - var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); - var allPageIds = locationsByPage.Keys.ToArray(); - var pageCount = allPageIds.Length; - if (degreeOfParallelism <= 0) degreeOfParallelism = Environment.ProcessorCount; var semaphore = new SemaphoreSlim(degreeOfParallelism); var tasks = new List>>(); var keyMap = _storage.GetKeyReverseMap(); + const int batchSize = 128; - for (int pageIdx = 0; pageIdx < pageCount; pageIdx++) + foreach (var batch in _primaryIndex + .Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId) + .Select(entry => entry.Location) + .Chunk(batchSize)) { await semaphore.WaitAsync(ct); - var localPageId = allPageIds[pageIdx]; + var localBatch = batch; var task = Task.Run(async () => { try { - var buffer = ArrayPool.Shared.Rent(_storage.PageSize); var results = new List(); + var pageCache = new Dictionary(); try { - await _storage.ReadPageAsync(localPageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct); - - foreach (var location in locationsByPage[localPageId]) + foreach (var location in localBatch) { + ct.ThrowIfCancellationRequested(); + + if (!pageCache.TryGetValue(location.PageId, out var buffer)) + { + buffer = ArrayPool.Shared.Rent(_storage.PageSize); + await _storage.ReadPageAsync(location.PageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); + pageCache[location.PageId] = buffer; + } + if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) { var rawBytes = ReadRawBytesAt(location, txnId, buffer); @@ -1343,7 +1350,8 @@ public async IAsyncEnumerable ParallelScanAsync( } finally { - ArrayPool.Shared.Return(buffer); + foreach (var buffer in pageCache.Values) + ArrayPool.Shared.Return(buffer); } return results; @@ -1355,15 +1363,27 @@ public async IAsyncEnumerable ParallelScanAsync( }, ct); tasks.Add(task); + + if (tasks.Count >= degreeOfParallelism) + { + var completedTask = await Task.WhenAny(tasks).ConfigureAwait(false); + tasks.Remove(completedTask); + + var results = await completedTask.ConfigureAwait(false); + foreach (var doc in results) + { + yield return doc; + } + } } // Yield results as tasks complete while (tasks.Count > 0) { - var completedTask = await Task.WhenAny(tasks); + var completedTask = await Task.WhenAny(tasks).ConfigureAwait(false); tasks.Remove(completedTask); - var results = await completedTask; + var results = await completedTask.ConfigureAwait(false); foreach (var doc in results) { yield return doc; From 1faf0f4dda0c0124c1ebde36625cdc37d9e7ffc7 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 14:43:18 +0000 Subject: [PATCH 4/4] fix: stream BSON scans from primary index Co-authored-by: mrdevrobot <12503462+mrdevrobot@users.noreply.github.com> --- .../Collections/DocumentCollection.cs | 269 +++++++++++------- .../CrossCollectionQueryIsolationTests.cs | 19 +- tests/BLite.Tests/RetentionPolicyTests.cs | 28 ++ 3 files changed, 197 insertions(+), 119 deletions(-) diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 532e24da..a721d9ee 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -352,7 +352,7 @@ private async Task ApplyRetentionPolicyCoreAsync(CancellationToken ct) /// /// Reads the raw BSON bytes for a document at the given location without deserializing. /// Returns null if the slot is deleted or the page/location is invalid. - /// Overflow documents are reassembled into a single BSON payload when needed. + /// Overflow documents are reassembled only when is true. /// private static bool TryReadInlineRawBytes(byte[] pageBuffer, ushort slotIndex, out ReadOnlySpan rawBytes) { @@ -373,7 +373,7 @@ private static bool TryReadInlineRawBytes(byte[] pageBuffer, ushort slotIndex, o return true; } - private byte[]? ReadRawBytesAt(DocumentLocation location, ulong txnId, byte[]? preloadedPage = null) + private byte[]? ReadRawBytesAt(DocumentLocation location, ulong txnId, bool includeOverflow = false, byte[]? preloadedPage = null) { byte[]? ownedBuffer = null; var buffer = preloadedPage ?? (ownedBuffer = ArrayPool.Shared.Rent(_storage.PageSize)); @@ -396,6 +396,7 @@ private static bool TryReadInlineRawBytes(byte[] pageBuffer, ushort slotIndex, o var slot = SlotEntry.ReadFrom(buffer.AsSpan(slotOffset)); if ((slot.Flags & SlotFlags.Deleted) != 0) return null; if ((slot.Flags & SlotFlags.HasOverflow) == 0) return null; + if (!includeOverflow) return null; if (slot.Offset + slot.Length > buffer.Length || slot.Length < 8) return null; @@ -442,28 +443,132 @@ private static bool TryReadInlineRawBytes(byte[] pageBuffer, ushort slotIndex, o } } - private async Task>> GetCollectionLocationsByPageAsync( - ulong txnId, - CancellationToken ct = default) + private bool MatchesIndexEntryKey(ReadOnlySpan bsonBytes, IndexKey expectedKey) { - var locationsByPage = new Dictionary>(); + if (TryReadIndexKeyFromBson(bsonBytes, out var actualKey)) + return actualKey.Equals(expectedKey); - await foreach (var entry in _primaryIndex - .RangeAsync(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId, ct) - .ConfigureAwait(false)) + try { - ct.ThrowIfCancellationRequested(); + var entity = _mapper.Deserialize(new BsonSpanReader(bsonBytes, _storage.GetKeyReverseMap())); + return _mapper.ToIndexKey(_mapper.GetId(entity)).Equals(expectedKey); + } + catch + { + return false; + } + } + + private bool TryReadIndexKeyFromBson(ReadOnlySpan bsonBytes, out IndexKey key) + { + key = default; + try + { + var reader = new BsonSpanReader(bsonBytes, _storage.GetKeyReverseMap()); + reader.ReadDocumentSize(); + + if (_storage.GetKeyMap().TryGetValue("_id", out var idFieldId) && + reader.TrySeekToField(idFieldId, out var seekType)) + { + return TryReadIndexKeyFromReader(ref reader, seekType, out key); + } + + while (reader.Remaining > 1) + { + var type = reader.ReadBsonType(); + if (type == BsonType.EndOfDocument) break; + + var name = reader.ReadElementHeader(); + if (name == "_id") + return TryReadIndexKeyFromReader(ref reader, type, out key); + + reader.SkipValue(type); + } + } + catch + { + return false; + } + + return false; + } + + private static bool TryReadIndexKeyFromReader(ref BsonSpanReader reader, BsonType type, out IndexKey key) + { + key = default; + + if (typeof(TId) == typeof(ObjectId) && type == BsonType.ObjectId) + { + key = IndexKey.Create(reader.ReadObjectId()); + return true; + } - if (!locationsByPage.TryGetValue(entry.Location.PageId, out var locations)) + if (typeof(TId) == typeof(int) && type == BsonType.Int32) + { + key = IndexKey.Create(reader.ReadInt32()); + return true; + } + + if (typeof(TId) == typeof(long) && type == BsonType.Int64) + { + key = IndexKey.Create(reader.ReadInt64()); + return true; + } + + if (type == BsonType.String) + { + var value = reader.ReadString(); + + if (typeof(TId) == typeof(string)) { - locations = new List(); - locationsByPage[entry.Location.PageId] = locations; + key = IndexKey.Create(value); + return true; } - locations.Add(entry.Location); + if (typeof(TId) == typeof(Guid) && Guid.TryParse(value, out var guid)) + { + key = IndexKey.Create(guid); + return true; + } } - return locationsByPage; + return false; + } + + private async IAsyncEnumerable<(IndexEntry Entry, byte[] RawBytes)> EnumerateOwnedRawDocumentsAsync( + ulong txnId, + bool includeOverflow, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct = default) + { + var pageCache = new Dictionary(); + + try + { + await foreach (var entry in _primaryIndex + .RangeAsync(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId, ct) + .ConfigureAwait(false)) + { + ct.ThrowIfCancellationRequested(); + + if (!pageCache.TryGetValue(entry.Location.PageId, out var cachedBuffer)) + { + cachedBuffer = ArrayPool.Shared.Rent(_storage.PageSize); + _storage.ReadPage(entry.Location.PageId, txnId, cachedBuffer); + pageCache[entry.Location.PageId] = cachedBuffer; + } + + var rawBytes = ReadRawBytesAt(entry.Location, txnId, includeOverflow, cachedBuffer); + if (rawBytes is null) continue; + if (!MatchesIndexEntryKey(rawBytes, entry.Key)) continue; + + yield return (entry, rawBytes); + } + } + finally + { + foreach (var buf in pageCache.Values) + ArrayPool.Shared.Return(buf); + } } /// @@ -977,37 +1082,18 @@ public async IAsyncEnumerable ScanAsync( var sw = _storage.MetricsDispatcher != null ? ValueStopwatch.StartNew() : default; var txnId = transaction?.TransactionId ?? 0UL; - var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); - var buffer = ArrayPool.Shared.Rent(_storage.PageSize); var keyMap = _storage.GetKeyReverseMap(); try { - foreach (var (pageId, locations) in locationsByPage) + await foreach (var (_, rawBytes) in EnumerateOwnedRawDocumentsAsync(txnId, includeOverflow: true, ct).ConfigureAwait(false)) { - ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - - foreach (var location in locations) - { - if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) - { - var rawBytes = ReadRawBytesAt(location, txnId, buffer); - if (rawBytes is null) continue; - inlineRawBytes = rawBytes; - } - - var reader = new BsonSpanReader(inlineRawBytes, keyMap); - if (!predicate(reader)) continue; - - var doc = await FindByLocationAsync(location, txnId, buffer, ct).ConfigureAwait(false); - if (doc != null) yield return doc; - } + if (!predicate(new BsonSpanReader(rawBytes, keyMap))) continue; + yield return _mapper.Deserialize(new BsonSpanReader(rawBytes, keyMap)); } } finally { - ArrayPool.Shared.Return(buffer); if (sw.IsActive) _storage.MetricsDispatcher?.Publish(new MetricEvent { @@ -1163,36 +1249,12 @@ public async IAsyncEnumerable ScanAsync( if (projector == null) throw new ArgumentNullException(nameof(projector)); var txnId = 0UL; - var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); - var buffer = ArrayPool.Shared.Rent(_storage.PageSize); var keyMap = _storage.GetKeyReverseMap(); - try - { - foreach (var (pageId, locations) in locationsByPage) - { - ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - - foreach (var location in locations) - { - if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) - { - var rawBytes = ReadRawBytesAt(location, txnId, buffer); - if (rawBytes is null) continue; - inlineRawBytes = rawBytes; - } - - var result = projector(new BsonSpanReader(inlineRawBytes, keyMap)); - if (result is null) continue; - - yield return result; - } - } - } - finally + await foreach (var (_, rawBytes) in EnumerateOwnedRawDocumentsAsync(txnId, includeOverflow: true, ct).ConfigureAwait(false)) { - ArrayPool.Shared.Return(buffer); + var result = projector(new BsonSpanReader(rawBytes, keyMap)); + if (result is not null) yield return result; } } @@ -1212,41 +1274,19 @@ internal async Task CountScanAsync( if (predicate == null) throw new ArgumentNullException(nameof(predicate)); var txnId = 0UL; - var locationsByPage = await GetCollectionLocationsByPageAsync(txnId, ct).ConfigureAwait(false); - var buffer = ArrayPool.Shared.Rent(_storage.PageSize); int count = 0; var keyMap = _storage.GetKeyReverseMap(); - try + await foreach (var (_, rawBytes) in EnumerateOwnedRawDocumentsAsync(txnId, includeOverflow: true, ct).ConfigureAwait(false)) { - foreach (var (pageId, locations) in locationsByPage) + try { - ct.ThrowIfCancellationRequested(); - await _storage.ReadPageAsync(pageId, txnId, buffer.AsMemory(0, _storage.PageSize), ct).ConfigureAwait(false); - - foreach (var location in locations) - { - if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) - { - var rawBytes = ReadRawBytesAt(location, txnId, buffer); - if (rawBytes is null) continue; - inlineRawBytes = rawBytes; - } - - try - { - if (predicate(new BsonSpanReader(inlineRawBytes, keyMap))) count++; - } - catch - { - // Malformed BSON — skip silently. - } - } + if (predicate(new BsonSpanReader(rawBytes, keyMap))) count++; + } + catch + { + // Malformed BSON — skip silently. } - } - finally - { - ArrayPool.Shared.Return(buffer); } return count; @@ -1305,14 +1345,12 @@ public async IAsyncEnumerable ParallelScanAsync( var tasks = new List>>(); var keyMap = _storage.GetKeyReverseMap(); const int batchSize = 128; + var batch = new List(batchSize); - foreach (var batch in _primaryIndex - .Range(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId) - .Select(entry => entry.Location) - .Chunk(batchSize)) + async Task QueueBatchAsync(List entries) { - await semaphore.WaitAsync(ct); - var localBatch = batch; + await semaphore.WaitAsync(ct).ConfigureAwait(false); + var localBatch = entries.ToArray(); var task = Task.Run(async () => { @@ -1323,9 +1361,10 @@ public async IAsyncEnumerable ParallelScanAsync( try { - foreach (var location in localBatch) + foreach (var entry in localBatch) { ct.ThrowIfCancellationRequested(); + var location = entry.Location; if (!pageCache.TryGetValue(location.PageId, out var buffer)) { @@ -1334,18 +1373,13 @@ public async IAsyncEnumerable ParallelScanAsync( pageCache[location.PageId] = buffer; } - if (!TryReadInlineRawBytes(buffer, location.SlotIndex, out var inlineRawBytes)) - { - var rawBytes = ReadRawBytesAt(location, txnId, buffer); - if (rawBytes is null) continue; - inlineRawBytes = rawBytes; - } - - if (!predicate(new BsonSpanReader(inlineRawBytes, keyMap))) + var rawBytes = ReadRawBytesAt(location, txnId, includeOverflow: true, buffer); + if (rawBytes is null) continue; + if (!MatchesIndexEntryKey(rawBytes, entry.Key)) continue; + if (!predicate(new BsonSpanReader(rawBytes, keyMap))) continue; - var doc = await FindByLocationAsync(location, txnId, buffer, ct).ConfigureAwait(false); - if (doc != null) results.Add(doc); + results.Add(_mapper.Deserialize(new BsonSpanReader(rawBytes, keyMap))); } } finally @@ -1363,7 +1397,17 @@ public async IAsyncEnumerable ParallelScanAsync( }, ct); tasks.Add(task); + } + await foreach (var entry in _primaryIndex + .RangeAsync(IndexKey.MinKey, IndexKey.MaxKey, IndexDirection.Forward, txnId, ct) + .ConfigureAwait(false)) + { + ct.ThrowIfCancellationRequested(); + batch.Add(entry); + if (batch.Count < batchSize) continue; + + await QueueBatchAsync(batch).ConfigureAwait(false); if (tasks.Count >= degreeOfParallelism) { var completedTask = await Task.WhenAny(tasks).ConfigureAwait(false); @@ -1375,8 +1419,13 @@ public async IAsyncEnumerable ParallelScanAsync( yield return doc; } } + + batch = new List(batchSize); } + if (batch.Count > 0) + await QueueBatchAsync(batch).ConfigureAwait(false); + // Yield results as tasks complete while (tasks.Count > 0) { diff --git a/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs b/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs index 1d952a7f..26a0a50e 100644 --- a/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs +++ b/tests/BLite.Tests/CrossCollectionQueryIsolationTests.cs @@ -87,28 +87,29 @@ public async Task Where_OnUnindexedSharedField_DoesNotReturnCrossCollectionRows( } [Fact] - public async Task Count_OnUnindexedSharedField_DoesNotCountCrossCollectionRows() + public async Task CountAsync_OnUnindexedNumericField_DoesNotCountCrossCollectionRows() { - await _db.IntEntities.InsertAsync(new IntEntity { Id = 1, Name = "Pranzo" }); - await _db.People.InsertAsync(new Person { Id = 101, Name = "Pranzo", Age = 20 }); + await _db.Users.InsertAsync(new User { Name = "Alpha", Age = 30 }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Foreign", Age = 30 }); await _db.SaveChangesAsync(); - var count = _db.IntEntities.AsQueryable() - .Count(x => x.Name == "Pranzo"); + var count = await _db.Users.AsQueryable() + .CountAsync(x => x.Age == 30); Assert.Equal(1, count); } [Fact] - public async Task Max_OnUnindexedSharedField_DoesNotReadCrossCollectionRows() + public async Task MaxAsync_FallbackScan_OnUnindexedNumericField_DoesNotReadCrossCollectionRows() { await _db.Users.InsertAsync(new User { Name = "Alpha", Age = 30 }); - await _db.People.InsertAsync(new Person { Id = 101, Name = "Zulu", Age = 20 }); + await _db.People.InsertAsync(new Person { Id = 101, Name = "Foreign", Age = 99 }); await _db.SaveChangesAsync(); - var maxName = _db.Users.AsQueryable().Max(x => x.Name); + var plan = IndexMinMax.Scan(BsonAggregator.Max("age")); + var maxAge = await _db.Users.AsQueryable().MaxAsync(plan); - Assert.Equal("Alpha", maxName); + Assert.Equal(30, maxAge); } [Fact] diff --git a/tests/BLite.Tests/RetentionPolicyTests.cs b/tests/BLite.Tests/RetentionPolicyTests.cs index 3d107e5d..445e1d3d 100644 --- a/tests/BLite.Tests/RetentionPolicyTests.cs +++ b/tests/BLite.Tests/RetentionPolicyTests.cs @@ -442,6 +442,34 @@ public async Task TypedCollection_MaxAge_DeletesOlderThanCutoff() Assert.Equal("new1", results[0].SensorId); } + [Fact] + public async Task TypedCollection_MaxAge_ExemptsOverflowDocuments() + { + using var col = GetCollection(); + col.SetRetentionPolicy(new RetentionPolicy + { + MaxAgeMs = (long)TimeSpan.FromHours(1).TotalMilliseconds, + TimestampField = "timestamp", + Triggers = RetentionTrigger.None + }); + + var old = DateTime.UtcNow.AddHours(-2); + var now = DateTime.UtcNow; + var overflowSensorId = new string('O', 20 * 1024); + + await col.InsertAsync(new BLite.Shared.SensorReading { SensorId = overflowSensorId, Value = 1, Timestamp = old }); + await col.InsertAsync(new BLite.Shared.SensorReading { SensorId = "old-inline", Value = 2, Timestamp = old }); + await col.InsertAsync(new BLite.Shared.SensorReading { SensorId = "new-inline", Value = 3, Timestamp = now }); + + await col.ForceApplyRetentionPolicyAsync(); + + var results = await col.FindAllAsync().ToListAsync(); + Assert.Equal(2, results.Count); + Assert.Contains(results, r => r.SensorId == overflowSensorId); + Assert.Contains(results, r => r.SensorId == "new-inline"); + Assert.DoesNotContain(results, r => r.SensorId == "old-inline"); + } + [Fact] public async Task TypedCollection_OnInsert_TriggersRetention() {