Merge pull request #40 from koculu/invalid-key-deletion-on-read-only-…

…memory-segment-loader

Fixed invalid key deletion bug on tree load.

src/ZoneTree.UnitTests/FixedSizeKeyAndValueTests.cs

@@ -120,10 +120,16 @@ public void IntStringGarbageCollectionTest()
             data.TryAtomicAdd(2, "2");
             data.TryAtomicAdd(3, "3");
             data.TryDelete(2);
+            data.TryAtomicAdd(4, "4");
+            data.TryAtomicUpdate(3, "33");
+            data.TryDelete(2);
             Assert.That(data.ContainsKey(1), Is.True);
             Assert.That(data.ContainsKey(2), Is.False);
             Assert.That(data.ContainsKey(3), Is.True);
-            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(3));
+            Assert.That(data.ContainsKey(4), Is.True);
+            data.TryGet(3, out var value3);
+            Assert.That(value3, Is.EqualTo("33"));
+            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(4));
         }
 
         // reload tree and check the length
@@ -136,7 +142,95 @@ public void IntStringGarbageCollectionTest()
             Assert.That(data.ContainsKey(1), Is.True);
             Assert.That(data.ContainsKey(2), Is.False);
             Assert.That(data.ContainsKey(3), Is.True);
-            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(2));
+            Assert.That(data.ContainsKey(4), Is.True);
+            data.TryGet(3, out var value3);
+            Assert.That(value3, Is.EqualTo("33"));
+            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(3));
+        }
+    }
+
+    [Test]
+    public void IntStringReadOnlySegmentLoadingTest()
+    {
+        var dataPath = "data/IntStringGarbageCollectionTest";
+        if (Directory.Exists(dataPath))
+            Directory.Delete(dataPath, true);
+
+        // load and populate tree
+        {
+            using var data = new ZoneTreeFactory<int, string>()
+                .SetDataDirectory(dataPath)
+                .OpenOrCreate();
+            data.TryAtomicAdd(1, "1");
+            data.TryAtomicAdd(2, "2");
+            data.TryAtomicAdd(3, "3");
+            data.TryDelete(2);
+            data.TryAtomicAdd(4, "4");
+            data.TryAtomicUpdate(3, "33");
+            data.TryDelete(2);
+            Assert.That(data.ContainsKey(1), Is.True);
+            Assert.That(data.ContainsKey(2), Is.False);
+            Assert.That(data.ContainsKey(3), Is.True);
+            Assert.That(data.ContainsKey(4), Is.True);
+            data.TryGet(3, out var value3);
+            Assert.That(value3, Is.EqualTo("33"));
+            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(4));
+            data.Maintenance.MoveMutableSegmentForward();
+            Assert.That(data.Maintenance.ReadOnlySegments[0].Length, Is.EqualTo(4));
+        }
+
+        // reload tree and check the length
+        for (var i = 0; i < 3; ++i)
+        {
+            using var data = new ZoneTreeFactory<int, string>()
+                .SetDataDirectory(dataPath)
+                .Open();
+            Assert.That(data.ContainsKey(1), Is.True);
+            Assert.That(data.ContainsKey(2), Is.False);
+            Assert.That(data.ContainsKey(3), Is.True);
+            Assert.That(data.ContainsKey(4), Is.True);
+            data.TryGet(3, out var value3);
+            Assert.That(value3, Is.EqualTo("33"));
+            Assert.That(data.Maintenance.ReadOnlySegments[0].Length, Is.EqualTo(4));
+        }
+    }
+
+    [Test]
+    public void IntStringDiskSegmentLoadingTest()
+    {
+        var dataPath = "data/IntStringGarbageCollectionTest";
+        if (Directory.Exists(dataPath))
+            Directory.Delete(dataPath, true);
+
+        // load and populate tree
+        {
+            using var data = new ZoneTreeFactory<int, string>()
+                .SetDataDirectory(dataPath)
+                .OpenOrCreate();
+            data.TryAtomicAdd(1, "1");
+            data.TryAtomicAdd(2, "2");
+            data.TryAtomicAdd(3, "3");
+            data.TryDelete(2);
+            Assert.That(data.ContainsKey(1), Is.True);
+            Assert.That(data.ContainsKey(2), Is.False);
+            Assert.That(data.ContainsKey(3), Is.True);
+            Assert.That(data.Maintenance.MutableSegment.Length, Is.EqualTo(3));
+            data.Maintenance.MoveMutableSegmentForward();
+            Assert.That(data.Maintenance.ReadOnlySegments[0].Length, Is.EqualTo(3));
+            data.Maintenance.StartMergeOperation().Join();
+            Assert.That(data.Maintenance.DiskSegment.Length, Is.EqualTo(2));
+        }
+
+        // reload tree and check the length
+        for (var i = 0; i < 3; ++i)
+        {
+            using var data = new ZoneTreeFactory<int, string>()
+                .SetDataDirectory(dataPath)
+                .Open();
+            Assert.That(data.ContainsKey(1), Is.True);
+            Assert.That(data.ContainsKey(2), Is.False);
+            Assert.That(data.ContainsKey(3), Is.True);
+            Assert.That(data.Maintenance.DiskSegment.Length, Is.EqualTo(2));
         }
     }
 

src/ZoneTree/Collections/DictionaryOfDictionaryWithWAL.cs

@@ -73,7 +73,7 @@ void LoadFromWriteAheadLog()
         var len = keys.Count;
         for (var i = 0; i < len; ++i)
         {
-            Upsert(keys[i], values[i].Value1, values[i].Value2);
+            UpsertWithoutWal(keys[i], values[i].Value1, values[i].Value2);
         }
     }
 
@@ -113,6 +113,22 @@ public bool Upsert(in TKey1 key1, in TKey2 key2, in TValue value)
         return false;
     }
 
+    bool UpsertWithoutWal(in TKey1 key1, in TKey2 key2, in TValue value)
+    {
+        if (Dictionary.TryGetValue(key1, out var dic))
+        {
+            dic.Remove(key2);
+            dic.Add(key2, value);
+            return true;
+        }
+        dic = new Dictionary<TKey2, TValue>
+        {
+            { key2, value }
+        };
+        Dictionary[key1] = dic;
+        return false;
+    }
+
     long NextOpIndex()
     {
         return IdProvider.NextId();

src/ZoneTree/Collections/DictionaryWithWal.cs

@@ -70,7 +70,7 @@ public DictionaryWithWAL(
 
     void LoadFromWriteAheadLog()
     {
-        var result = WriteAheadLog.ReadLogEntries(false, false, false);
+        var result = WriteAheadLog.ReadLogEntries(false, false, true);
         if (!result.Success)
         {
             if (result.HasFoundIncompleteTailRecord)
@@ -87,7 +87,7 @@ void LoadFromWriteAheadLog()
         }
 
         (var newKeys, var newValues) = WriteAheadLogUtility
-            .StableSortAndCleanUpDeletedKeys(
+            .StableSortAndCleanUpDeletedAndDuplicatedKeys(
             result.Keys,
             result.Values,
             Comparer,

src/ZoneTree/Core/ZoneTreeLoader.cs

@@ -244,16 +244,19 @@ public ZoneTree<TKey, TValue> LoadZoneTree()
         if (collectGarbage)
         {
             var len = MutableSegment.Length;
-            var keys = new TKey[len];
-            var values = new TValue[len];
-            var iterator = MutableSegment.GetSeekableIterator();
-            var i = 0;
-            while (iterator.Next())
+            if (mutableSegmentWal.InitialLength != MutableSegment.Length)
             {
-                keys[i] = iterator.CurrentKey;
-                values[i++] = iterator.CurrentValue;
+                var keys = new TKey[len];
+                var values = new TValue[len];
+                var iterator = MutableSegment.GetSeekableIterator();
+                var i = 0;
+                while (iterator.Next())
+                {
+                    keys[i] = iterator.CurrentKey;
+                    values[i++] = iterator.CurrentValue;
+                }
+                mutableSegmentWal.ReplaceWriteAheadLog(keys, values, true);
             }
-            mutableSegmentWal.ReplaceWriteAheadLog(keys, values, true);
         }
         LoadDiskSegment();
         LoadBottomSegments();

src/ZoneTree/Segments/InMemory/ReadOnlySegmentLoader.cs

@@ -46,11 +46,10 @@ public IReadOnlySegment<TKey, TValue> LoadReadOnlySegment(long segmentId)
             ZoneTree<TKey, TValue>.SegmentWalCategory);
 
         (var newKeys, var newValues) = WriteAheadLogUtility
-            .StableSortAndCleanUpDeletedKeys(
+            .StableSortAndCleanUpDuplicatedKeys(
             result.Keys,
             result.Values,
-            Options.Comparer,
-            Options.IsValueDeleted);
+            Options.Comparer);
 
         var ros = new ReadOnlySegment<TKey, TValue>(
             segmentId,

src/ZoneTree/WAL/Async/AsyncCompressedFileSystemWriteAheadLog.cs

@@ -83,6 +83,8 @@ public override int GetHashCode()
 
     public bool EnableIncrementalBackup { get; set; }
 
+    public int InitialLength { get; private set; }
+
     public AsyncCompressedFileSystemWriteAheadLog(
         ILogger logger,
         IFileStreamProvider fileStreamProvider,
@@ -222,14 +224,16 @@ public WriteAheadLogReadLogEntriesResult<TKey, TValue> ReadLogEntries(
         bool stopReadOnChecksumFailure,
         bool sortByOpIndexes)
     {
-        return WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
+        var result = WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
             Logger,
             FileStream,
             stopReadOnException,
             stopReadOnChecksumFailure,
             LogEntry.ReadLogEntry,
             DeserializeLogEntry,
             sortByOpIndexes);
+        InitialLength = result.Keys.Count;
+        return result;
     }
 
     (bool isValid, TKey key, TValue value, long opIndex) DeserializeLogEntry(in LogEntry logEntry)

src/ZoneTree/WAL/IWriteAheadLog.cs

@@ -8,6 +8,12 @@ public interface IWriteAheadLog<TKey, TValue> : IDisposable
 
     bool EnableIncrementalBackup { get; set; }
 
+    /// <summary>
+    /// The initial record count of the log.
+    /// It is available after the ReadLogEntries call.
+    /// </summary>
+    int InitialLength { get; }
+
     void Append(in TKey key, in TValue value, long opIndex);
 
     void Drop();
@@ -27,13 +33,13 @@ WriteAheadLogReadLogEntriesResult<TKey, TValue> ReadLogEntries(
     /// <param name="disableBackup">disable backup regardless of wal flag.</param>
     /// <returns>the difference: old file length - new file length.</returns>
     long ReplaceWriteAheadLog(TKey[] keys, TValue[] values, bool disableBackup);
-    
+
     /// <summary>
     /// Informs the write ahead log that no further writes will be sent.
     /// to let WAL optimize itself.
     /// </summary>
     void MarkFrozen();
-    
+
     /// <summary>
     /// Truncates incomplete tail record.
     /// </summary>

src/ZoneTree/WAL/Null/NullWriteAheadLog.cs

@@ -8,6 +8,8 @@ public sealed class NullWriteAheadLog<TKey, TValue> : IWriteAheadLog<TKey, TValu
 
     public bool EnableIncrementalBackup { get; set; }
 
+    public int InitialLength { get; private set; }
+
     public void Append(in TKey key, in TValue value, long opIndex)
     {
     }

src/ZoneTree/WAL/Sync/SyncFileSystemWriteAheadLog.cs

@@ -33,6 +33,8 @@ public sealed class SyncFileSystemWriteAheadLog<TKey, TValue> : IWriteAheadLog<T
 
     public bool EnableIncrementalBackup { get; set; }
 
+    public int InitialLength { get; private set; }
+
     public SyncFileSystemWriteAheadLog(
         ILogger logger,
         IFileStreamProvider fileStreamProvider,
@@ -88,14 +90,16 @@ public WriteAheadLogReadLogEntriesResult<TKey, TValue> ReadLogEntries(
         bool stopReadOnChecksumFailure,
         bool sortByOpIndexes)
     {
-        return WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
+        var result = WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
             Logger,
             FileStream.ToStream(),
             stopReadOnException,
             stopReadOnChecksumFailure,
             LogEntry.ReadLogEntry,
             DeserializeLogEntry,
             sortByOpIndexes);
+        InitialLength = result.Keys.Count;
+        return result;
     }
 
     (bool isValid, TKey key, TValue value, long opIndex) DeserializeLogEntry(in LogEntry logEntry)

src/ZoneTree/WAL/SyncCompressed/SyncCompressedFileSystemWriteAheadLog.cs

@@ -38,6 +38,8 @@ public sealed class SyncCompressedFileSystemWriteAheadLog<TKey, TValue> : IWrite
 
     public bool EnableIncrementalBackup { get; set; }
 
+    public int InitialLength { get; private set; }
+
     public SyncCompressedFileSystemWriteAheadLog(
         ILogger logger,
         IFileStreamProvider fileStreamProvider,
@@ -98,14 +100,16 @@ public WriteAheadLogReadLogEntriesResult<TKey, TValue> ReadLogEntries(
         bool stopReadOnChecksumFailure,
         bool sortByOpIndexes)
     {
-        return WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
+        var result = WriteAheadLogEntryReader.ReadLogEntries<TKey, TValue, LogEntry>(
             Logger,
             FileStream,
             stopReadOnException,
             stopReadOnChecksumFailure,
             LogEntry.ReadLogEntry,
             DeserializeLogEntry,
             sortByOpIndexes);
+        InitialLength = result.Keys.Count;
+        return result;
     }
 
     (bool isValid, TKey key, TValue value, long opIndex) DeserializeLogEntry(in LogEntry logEntry)