diff --git a/src/SharpCoreDB/DataStructures/Table.CRUD.cs b/src/SharpCoreDB/DataStructures/Table.CRUD.cs index 1d2984c5..df21f7eb 100644 --- a/src/SharpCoreDB/DataStructures/Table.CRUD.cs +++ b/src/SharpCoreDB/DataStructures/Table.CRUD.cs @@ -2,6 +2,7 @@ namespace SharpCoreDB.DataStructures; using System; using System.Collections.Generic; +using System.Globalization; using System.Runtime.CompilerServices; using System.Threading; using System.Buffers; @@ -1931,6 +1932,16 @@ internal void UpdateMultiple(List<(string where, Dictionary upda // remove the stale record from all indexes (unloaded indexes would later be rebuilt // from the file INCLUDING the stale record). EnsureAllRegisteredIndexesLoaded(); + + // B8: single-pass contiguous UPDATE — when every operation is a `pk = ` match on a + // plaintext fixed-width table with physically adjacent PK-ordered records, the old records + // are read as ONE contiguous byte range and patched in memory (no per-row pread). Strictly + // gated; any mismatch falls back to the generic per-row loop below. + if (TryBulkUpdateContiguousFixedWidth(engine, operations)) + { + return; + } + int appendedInBatch = 0; // only appends create stale versions that need compaction foreach (var (where, updates) in operations) @@ -2316,6 +2327,222 @@ oldHashValues is not null && } } + /// + /// B8: number of UPDATE batches processed by the contiguous single-pass fast path (diagnostics + /// used by tests to prove the path engages; zero means every batch fell back to the generic loop). + /// + public long BulkContiguousUpdateBatches => Interlocked.Read(ref _bulkContiguousUpdateBatches); + + private long _bulkContiguousUpdateBatches; + + /// + /// B8: single-pass contiguous UPDATE fast path for plaintext fixed-width tables. Requires every + /// operation to be a simple pk = <numeric literal> match on the primary key, with keys + /// strictly increasing AND records physically adjacent in the data file (the fixed-width layout + /// makes the on-disk stride constant: [4-byte length][FixedSize payload]). When all + /// conditions hold the target records are read as one contiguous byte range, patched in memory + /// (field-level in-place patch, exactly like the generic row path) and written back through the + /// same buffered same-length overwrite. Any mismatch returns so the caller + /// falls back to the generic per-row loop — no records are touched before the range is verified. + /// + private bool TryBulkUpdateContiguousFixedWidth( + IStorageEngine engine, + List<(string where, Dictionary updates)> operations) + { + int count = operations.Count; + if (count < 2) + { + return false; + } + + // Narrow, conservative gate: fixed-width columnar table with an explicit PK, plaintext + // records only (a raw contiguous read must equal the logical record bytes), no buffered + // overwrites for this file, and no CHECK constraints (mirrors the generic fastPatch gate). + if (!_fixedWidthRecords || + StorageMode != StorageMode.Columnar || + this.PrimaryKeyIndex < 0 || + this.TableCheckConstraints.Count > 0 || + HasColumnCheckConstraints() || + this.storage is null || + this._config is not { NoEncryptMode: true } || + this.storage.HasBufferedOverwrite(DataFile)) + { + return false; + } + + var pkName = this.Columns[this.PrimaryKeyIndex]; + var layout = GetFixedWidthLayout(); + long stride = 4L + layout.FixedSize; + + var positions = new long[count]; + var keys = new string[count]; + var repointColumns = new List?[count]; + long parsedPrev = 0; + + for (int i = 0; i < count; i++) + { + var (where, updates) = operations[i]; + if (string.IsNullOrEmpty(where) || + !TryParseSimpleWhereClause(where, out var whereCol, out var whereVal) || + !string.Equals(whereCol, pkName, StringComparison.OrdinalIgnoreCase) || + whereVal is null) + { + return false; + } + + var keyStr = whereVal.ToString(); + if (string.IsNullOrEmpty(keyStr)) + { + return false; + } + + foreach (var updateKey in updates.Keys) + { + // Resolve the column; unknown columns must fall back so the generic loop surfaces the error. + int colIdx = -1; + for (int c = 0; c < this.Columns.Count; c++) + { + if (this.Columns[c].Equals(updateKey, StringComparison.OrdinalIgnoreCase)) + { + colIdx = c; + break; + } + } + + if (colIdx < 0) + { + return false; + } + + // Updating the PK itself must fall back (the PK B-tree position must be re-pointed). + if (colIdx == this.PrimaryKeyIndex) + { + return false; + } + + // Hash-indexed SET columns need their entries re-pointed after the in-place write. + // Fixed-size values decode from the raw slot cheaply; variable-length (TEXT/BLOB) + // indexed columns would need an overflow-arena read, so they fall back to the + // generic per-row loop. + if (this.hashIndexes.ContainsKey(this.Columns[colIdx])) + { + if (layout.IsVariable[colIdx]) + { + return false; + } + + (repointColumns[i] ??= new List(2)).Add(colIdx); + } + } + + // Strictly increasing numeric keys keep the physical records adjacent for fixed-width. + if (!long.TryParse(keyStr, NumberStyles.Integer, CultureInfo.InvariantCulture, out long key)) + { + return false; + } + + if (i > 0 && key <= parsedPrev) + { + return false; + } + + parsedPrev = key; + keys[i] = keyStr; + } + + // Resolve the first record position through the PK B-tree, then require every later key to + // sit at the expected contiguous offset (physical adjacency). Verifying every position keeps + // the gate sound even when earlier appends/deletes shifted records. + var first = this.Index.Search(keys[0]); + if (!first.Found) + { + return false; + } + + long basePosition = first.Value; + positions[0] = basePosition; + long expected = basePosition; + for (int i = 1; i < count; i++) + { + expected += stride; + var search = this.Index.Search(keys[i]); + if (!search.Found || search.Value != expected) + { + return false; + } + + positions[i] = expected; + } + + // Read the whole contiguous span in ONE range read (plaintext records only — enforced above). + long totalBytes = stride * count; + if (totalBytes <= 0 || totalBytes > int.MaxValue) + { + return false; + } + + var raw = this.storage.ReadBytesRange(DataFile, basePosition, (int)totalBytes); + if (raw is null || raw.Length < totalBytes) + { + return false; + } + + // Verify every 4-byte length prefix matches the fixed record size BEFORE touching anything. + for (int i = 0; i < count; i++) + { + int prefix = BinaryPrimitives.ReadInt32LittleEndian(raw.AsSpan((int)(i * stride), 4)); + if (prefix != layout.FixedSize) + { + return false; + } + } + + // Patch and write each record in place (buffered by the storage layer; flushed at commit). + for (int i = 0; i < count; i++) + { + var payload = new byte[layout.FixedSize]; + raw.AsSpan((int)(i * stride) + 4, layout.FixedSize).CopyTo(payload); + + var patched = TryOverwriteFixedWidthInPlace(payload, operations[i].updates); + if (patched is null || !engine.TryUpdateInPlaceSameLength(Name, positions[i], patched)) + { + return false; + } + + // Re-point hash-index entries for every fixed-size indexed SET column (mirrors the + // generic fastPatch path: old value decoded from the pre-write record, new value at the + // same position). + var repoints = repointColumns[i]; + if (repoints is { Count: > 0 }) + { + foreach (var colIdx in repoints) + { + var colName = this.Columns[colIdx]; + if (!this.hashIndexes.TryGetValue(colName, out var hashIdx)) + { + continue; + } + + var slot = payload.AsSpan(layout.Offsets[colIdx], layout.SlotSizes[colIdx]); + var oldVal = ReadTypedValueFromSpan(slot, this.ColumnTypes[colIdx], out _); + if (oldVal is not null) + { + hashIdx.Remove(oldVal, positions[i]); + } + + var newVal = operations[i].updates[colName]; + if (newVal is not null) + { + hashIdx.Add(newVal, positions[i]); + } + } + } + } + + Interlocked.Increment(ref _bulkContiguousUpdateBatches); + return true; + } + /// /// True when any column carries a CHECK expression (the batch fast-patch path is disabled in /// that case because a CHECK may read non-updated columns). diff --git a/src/SharpCoreDB/Interfaces/IStorage.cs b/src/SharpCoreDB/Interfaces/IStorage.cs index 03b27a39..94b4a70d 100644 --- a/src/SharpCoreDB/Interfaces/IStorage.cs +++ b/src/SharpCoreDB/Interfaces/IStorage.cs @@ -113,6 +113,13 @@ public interface IStorage bool OverwriteRecordAtSameLength(string path, long offset, byte[] data) => OverwriteRecordAt(path, offset, data); + /// + /// True when an in-place overwrite is currently buffered for (the + /// transaction write-behind buffer overlays reads of those offsets). Callers that bypass the + /// per-record read path must check this first so they never read stale disk bytes. + /// + bool HasBufferedOverwrite(string path) => false; + /// /// Appends multiple binary data blocks to a file in a single batch operation (used for batch inserts). /// @@ -148,6 +155,14 @@ bool OverwriteRecordAtSameLength(string path, long offset, byte[] data) => /// The read data, or null if file does not exist or position is invalid. byte[]? ReadBytesAt(string path, long position, int maxLength, bool noEncrypt); + /// + /// Reads a raw contiguous byte range starting at using the storage + /// layer's cached file handle (no per-call handle open). Used by the fixed-width contiguous + /// UPDATE fast path, which only engages on plaintext files. Implementations that cannot serve a + /// raw range (encrypted layouts, mocks) return null so the caller falls back to per-record reads. + /// + byte[]? ReadBytesRange(string path, long offset, int length) => null; + /// /// Enumerates every record in a table data file, yielding the literal file offset of the /// 4-byte length prefix (the offset returned by ) together with the diff --git a/src/SharpCoreDB/Services/Storage.Append.cs b/src/SharpCoreDB/Services/Storage.Append.cs index 5e2c5281..07311664 100644 --- a/src/SharpCoreDB/Services/Storage.Append.cs +++ b/src/SharpCoreDB/Services/Storage.Append.cs @@ -556,6 +556,10 @@ public bool OverwriteRecordAtSameLength(string path, long offset, byte[] data) return BufferOrWriteOverwriteInPlace(path, offset, record); } + /// + public bool HasBufferedOverwrite(string path) => + !bufferedOverwrites.IsEmpty && bufferedOverwrites.ContainsKey(path); + /// [MethodImpl(MethodImplOptions.AggressiveOptimization)] public long[] AppendBytesMultiple(string path, List dataBlocks) diff --git a/src/SharpCoreDB/Services/Storage.PageCache.cs b/src/SharpCoreDB/Services/Storage.PageCache.cs index 72fb85e8..00a96bac 100644 --- a/src/SharpCoreDB/Services/Storage.PageCache.cs +++ b/src/SharpCoreDB/Services/Storage.PageCache.cs @@ -5,6 +5,7 @@ namespace SharpCoreDB.Services; +using Microsoft.Win32.SafeHandles; using System; using System.IO; using System.Runtime.CompilerServices; @@ -59,6 +60,48 @@ public partial class Storage return ReadBytesAt(path, position, maxLength, false); } + /// + public byte[]? ReadBytesRange(string path, long offset, int length) + { + if (length <= 0 || length > 512 * 1024 * 1024) + { + return null; + } + + SafeFileHandle handle; + try + { + handle = GetOrOpenReadHandle(path); + } + catch + { + _readHandleCache.TryRemove(path, out _); + try + { + handle = GetOrOpenReadHandle(path); + } + catch + { + return null; + } + } + + var buffer = new byte[length]; + int total = 0; + while (total < length) + { + int read = RandomAccess.Read(handle, buffer.AsSpan(total), offset + total); + if (read <= 0) + { + return null; + } + + total += read; + } + + return buffer; + } + /// /// Loads a page from disk into a byte array for caching. /// diff --git a/tests/SharpCoreDB.Tests/FixedWidthBulkUpdateTests.cs b/tests/SharpCoreDB.Tests/FixedWidthBulkUpdateTests.cs new file mode 100644 index 00000000..c669e785 --- /dev/null +++ b/tests/SharpCoreDB.Tests/FixedWidthBulkUpdateTests.cs @@ -0,0 +1,232 @@ +// +// Copyright (c) 2026 MPCoreDeveloper. All rights reserved. +// Licensed under the MIT License. See LICENSE file in the project root for full license information. +// +namespace SharpCoreDB.Tests; + +using Microsoft.Extensions.DependencyInjection; +using SharpCoreDB.DataStructures; +using SharpCoreDB.Interfaces; +using SharpCoreDB.Storage.Hybrid; +using System; +using System.Collections.Generic; +using System.Globalization; +using System.IO; +using Xunit; + +/// +/// B8: single-pass contiguous UPDATE fast path (Table.TryBulkUpdateContiguousFixedWidth). When a +/// batch UPDATE hits a plaintext fixed-width table with strictly ascending `pk = literal` matches +/// whose records are physically adjacent, the target records are read as one contiguous byte range +/// and patched in memory. Every other shape must fall back to the generic per-row loop and stay +/// correct. +/// +public sealed class FixedWidthBulkUpdateTests : IDisposable +{ + private readonly DatabaseFactory _factory; + private readonly string _dirPath; + + public FixedWidthBulkUpdateTests() + { + var services = new ServiceCollection(); + services.AddSharpCoreDB(); + _factory = services.BuildServiceProvider().GetRequiredService(); + _dirPath = Path.Combine(Path.GetTempPath(), $"SCDB_BulkUpd_{Guid.NewGuid():N}"); + } + + public void Dispose() + { + try { if (Directory.Exists(_dirPath)) Directory.Delete(_dirPath, true); } catch { } + } + + private IDatabase CreateDb(bool noEncrypt = true) => _factory.Create(_dirPath, "pw", isReadOnly: false, + config: new DatabaseConfig { NoEncryptMode = noEncrypt }); + + private static Table TableOf(IDatabase db, string tableName) + { + Assert.True(db.TryGetTable(tableName, out var t)); + return Assert.IsType(t); + } + + private static void InsertDocs(IDatabase db, int fromId, int toId) + { + var stmts = new List(toId - fromId + 1); + for (int i = fromId; i <= toId; i++) + { + stmts.Add(string.Format(CultureInfo.InvariantCulture, + "INSERT INTO docs VALUES ({0}, 'user{0}', {1}, {2})", i, i * 0.5, 20 + (i % 60))); + } + + db.ExecuteBatchSQL(stmts); + } + + private static List BuildUpdates(int fromId, int toId, string setExpr) + { + var stmts = new List(Math.Abs(toId - fromId) + 1); + for (int i = fromId; i <= toId; i++) + { + stmts.Add(string.Format(CultureInfo.InvariantCulture, + "UPDATE docs SET {0} WHERE id = {1}", setExpr, i)); + } + + return stmts; + } + + [Fact] + public void ContiguousAscendingUpdates_EngageBulkPath_AndPersist() + { + IDatabase? db = null; + try + { + db = CreateDb(); + db.ExecuteSQL("CREATE TABLE docs (id INTEGER PRIMARY KEY, name TEXT, score REAL, age INTEGER)"); + InsertDocs(db, 1, 2000); + db.Flush(); + + var table = TableOf(db, "docs"); + Assert.True(table.IsFixedWidthRecords); + Assert.Equal(StorageMode.Columnar, table.StorageMode); + Assert.Equal(0, table.BulkContiguousUpdateBatches); + + // 1000 ascending `id = literal` updates — every row patched via one contiguous read. + db.ExecuteBatchSQL(BuildUpdates(1, 1000, "score = 99.0")); + db.Flush(); + Assert.Equal(1, table.BulkContiguousUpdateBatches); + + // Full-scan integrity: no rows lost, untouched rows unchanged. + var all = db.ExecuteQuery("SELECT COUNT(*) AS total FROM docs"); + Assert.Equal(2000L, Convert.ToInt64(all[0].Values.First())); + var tail = db.ExecuteQuery("SELECT score FROM docs WHERE id = 2000"); + Assert.Equal(1000.0, Convert.ToDouble(tail[0]["score"])); + + // Reopen: the in-place writes must be durable and readable. + (db as IDisposable)?.Dispose(); + db = CreateDb(); + for (int i = 1; i <= 2000; i += 251) + { + var rows = db.ExecuteQuery("SELECT score FROM docs WHERE id = @id", + new Dictionary { ["@id"] = i }); + Assert.Single(rows); + Assert.Equal(i <= 1000 ? 99.0 : i * 0.5, Convert.ToDouble(rows[0]["score"])); + } + } + finally { (db as IDisposable)?.Dispose(); } + } + + [Fact] + public void GappedUpdates_FallBackToGenericLoop_AndStayCorrect() + { + var db = CreateDb(); + try + { + db.ExecuteSQL("CREATE TABLE docs (id INTEGER PRIMARY KEY, name TEXT, score REAL)"); + InsertDocs(db, 1, 1000); + db.Flush(); + + var table = TableOf(db, "docs"); + + // Odd ids only: the records are not physically adjacent within the batch (id 1,3,5,...), + // so the contiguous fast path must refuse and the generic per-row loop applies them. + var stmts = new List(500); + for (int i = 1; i <= 1000; i += 2) + { + stmts.Add(string.Format(CultureInfo.InvariantCulture, "UPDATE docs SET score = 7.5 WHERE id = {0}", i)); + } + + db.ExecuteBatchSQL(stmts); + db.Flush(); + Assert.Equal(0, table.BulkContiguousUpdateBatches); + + for (int i = 1; i <= 1000; i += 137) + { + var rows = db.ExecuteQuery("SELECT score FROM docs WHERE id = @id", + new Dictionary { ["@id"] = i }); + Assert.Single(rows); + Assert.Equal(i % 2 == 1 ? 7.5 : i * 0.5, Convert.ToDouble(rows[0]["score"])); + } + } + finally { (db as IDisposable)?.Dispose(); } + } + + [Fact] + public void RepeatedCommits_EngageAgain_WithCurrentBytes() + { + var db = CreateDb(); + try + { + db.ExecuteSQL("CREATE TABLE docs (id INTEGER PRIMARY KEY, name TEXT, score REAL)"); + InsertDocs(db, 1, 500); + db.Flush(); + + var table = TableOf(db, "docs"); + db.ExecuteBatchSQL(BuildUpdates(1, 500, "score = 1.0")); + db.Flush(); + Assert.Equal(1, table.BulkContiguousUpdateBatches); + + // Second committed batch over the same rows must see the first batch's bytes (no stale reads). + db.ExecuteBatchSQL(BuildUpdates(1, 500, "score = 2.0")); + db.Flush(); + Assert.Equal(2, table.BulkContiguousUpdateBatches); + + var rows = db.ExecuteQuery("SELECT score FROM docs WHERE id = 250"); + Assert.Single(rows); + Assert.Equal(2.0, Convert.ToDouble(rows[0]["score"])); + } + finally { (db as IDisposable)?.Dispose(); } + } + + [Fact] + public void IndexedColumnUpdate_FallsBack_AndRepointsIndex() + { + var db = CreateDb(); + try + { + db.ExecuteSQL("CREATE TABLE docs (id INTEGER PRIMARY KEY, name TEXT, score REAL)"); + db.ExecuteSQL("CREATE INDEX idx_docs_name ON docs(name)"); + InsertDocs(db, 1, 800); + db.Flush(); + + var table = TableOf(db, "docs"); + // name is hash-indexed → the bulk path must refuse (index re-point is required). + var stmts = new List(800); + for (int i = 1; i <= 800; i++) + { + stmts.Add(string.Format(CultureInfo.InvariantCulture, + "UPDATE docs SET name = 'renamed-{0}' WHERE id = {0}", i)); + } + + db.ExecuteBatchSQL(stmts); + Assert.Equal(0, table.BulkContiguousUpdateBatches); + + // Index lookups must return the re-pointed rows (no stale entries). + var byIndex = db.ExecuteQuery("SELECT id FROM docs WHERE name = 'renamed-42'"); + Assert.Single(byIndex); + Assert.Equal(42L, Convert.ToInt64(byIndex[0]["id"])); + var byPk = db.ExecuteQuery("SELECT name FROM docs WHERE id = 42"); + Assert.Equal("renamed-42", byPk[0]["name"]); + } + finally { (db as IDisposable)?.Dispose(); } + } + + [Fact] + public void EncryptedOrDefaultConfig_FallsBack_AndStaysCorrect() + { + var db = CreateDb(noEncrypt: false); // default config: the gate requires NoEncryptMode + try + { + db.ExecuteSQL("CREATE TABLE docs (id INTEGER PRIMARY KEY, name TEXT, score REAL)"); + InsertDocs(db, 1, 300); + db.Flush(); + + var table = TableOf(db, "docs"); + db.ExecuteBatchSQL(BuildUpdates(1, 300, "score = 4.25")); + db.Flush(); + Assert.Equal(0, table.BulkContiguousUpdateBatches); + + var rows = db.ExecuteQuery("SELECT score FROM docs WHERE id = 300"); + Assert.Single(rows); + Assert.Equal(4.25, Convert.ToDouble(rows[0]["score"])); + } + finally { (db as IDisposable)?.Dispose(); } + } +}