STREAMS API in Garnet

Garnet.sln

@@ -113,6 +113,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NoOpModule", "playground\No
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ETag", "samples\ETag\ETag.csproj", "{4FBA1587-BAFC-49F8-803A-D1CF431A26F5}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Btree", "playground\BTree\Btree.csproj", "{CE12831B-2805-469E-8208-759DC4B4862C}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -345,6 +347,14 @@ Global
 		{4FBA1587-BAFC-49F8-803A-D1CF431A26F5}.Release|Any CPU.Build.0 = Release|Any CPU
 		{4FBA1587-BAFC-49F8-803A-D1CF431A26F5}.Release|x64.ActiveCfg = Release|Any CPU
 		{4FBA1587-BAFC-49F8-803A-D1CF431A26F5}.Release|x64.Build.0 = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|x64.Build.0 = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|Any CPU.Build.0 = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|x64.ActiveCfg = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|x64.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -381,6 +391,7 @@ Global
 		{A48412B4-FD60-467E-A5D9-F155CAB4F907} = {147FCE31-EC09-4C90-8E4D-37CA87ED18C3}
 		{D4C9A1A0-7053-F072-21F5-4E0C5827136D} = {69A71E2C-00E3-42F3-854E-BE157A24834E}
 		{4FBA1587-BAFC-49F8-803A-D1CF431A26F5} = {7068BB97-1958-4060-B5F1-859464592E56}
+		{CE12831B-2805-469E-8208-759DC4B4862C} = {69A71E2C-00E3-42F3-854E-BE157A24834E}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {2C02C405-4798-41CA-AF98-61EDFEF6772E}

libs/common/RespReadUtils.cs

@@ -630,6 +630,16 @@ public static bool TryReadInt64WithLengthHeader(out long number, ref byte* ptr,
             return true;
         }
 
+         /// <summary>
+        /// Tries to read a Ulong from the given ASCII-encoded RESP string. 
+        /// Note: this does not check for any length headers and is simply an accessor to TryReadUlong.
+        /// </summary>
+        /// <param name="number">If parsing was successful, contains the parsed ulong value.</param>
+        /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
+        /// <param name="end">The current end of the RESP string.</param>
+        /// <returns>True if a ulong was successfully parsed.</returns>
+        public static bool ReadUlong(out ulong number, ref byte* ptr, byte* end) => TryReadUInt64(ref ptr, end, out number, out _);
+
         /// <summary>
         /// Read long with length header
         /// </summary>

libs/resources/RespCommandsInfo.json

@@ -4828,6 +4828,124 @@
       }
     ]
   },
+  {
+    "Command": "XADD",
+    "Name": "XADD",
+    "IsInternal": false,
+    "Arity": -5,
+    "Flags": "DenyOom, Fast, Write",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Stream, Write",
+    "Tips": [
+      "nondeterministic_output"
+    ],
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": "UPDATE instead of INSERT because of the optional trimming feature",
+        "Flags": "RW, Update"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XDEL",
+    "Name": "XDEL",
+    "IsInternal": false,
+    "Arity": -3,
+    "Flags": "Fast, Write",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Stream, Write",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RW, Delete"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XLEN",
+    "Name": "XLEN",
+    "IsInternal": false,
+    "Arity": 2,
+    "Flags": "Fast, ReadOnly",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Read, Stream",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RO"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XRANGE",
+    "Name": "XRANGE",
+    "IsInternal": false,
+    "Arity": -4,
+    "Flags": "ReadOnly",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Read, Slow, Stream",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RO, Access"
+      }
+    ],
+    "SubCommands": null
+  },
   {
     "Command": "ZADD",
     "Name": "ZADD",

libs/server/BTreeIndex/BTree.cs

@@ -0,0 +1,177 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using System;
+using System.Collections.Generic;
+using System.Runtime.InteropServices;
+
+namespace Garnet.server.BTreeIndex
+{
+    public unsafe partial class BTree
+    {
+        BTreeNode* root;
+        BTreeNode* head;
+        BTreeNode* tail;
+        byte* tailMinKey;
+        public static readonly int MAX_TREE_DEPTH = 10; // maximum allowed depth of the tree
+        static int DEFAULT_SPLIT_LEAF_POSITION = (BTreeNode.LEAF_CAPACITY + 1) / 2; // position at which leaf node is split
+        static int SPLIT_LEAF_POSITION = BTreeNode.LEAF_CAPACITY; // position at which leaf node is split
+        static int SPLIT_INTERNAL_POSITION = BTreeNode.INTERNAL_CAPACITY; // position at which internal node is split
+
+        BTreeNode*[] rootToTailLeaf; // array of nodes from root to tail leaf
+        public BTreeStats stats; // statistics about the tree
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="BTree"/> class.
+        /// </summary>
+        public BTree(uint sectorSize)
+        {
+            var memoryBlock = (IntPtr*)NativeMemory.AlignedAlloc((nuint)BTreeNode.PAGE_SIZE, (nuint)BTreeNode.PAGE_SIZE);
+            root = BTreeNode.Create(BTreeNodeType.Leaf, memoryBlock);
+            head = tail = root;
+            root->info->next = root->info->previous = null;
+            root->info->count = 0;
+            tailMinKey = null;
+            rootToTailLeaf = new BTreeNode*[MAX_TREE_DEPTH];
+            stats = new BTreeStats();
+            stats.depth = 1;
+            stats.numLeafNodes = 1;
+            stats.numAllocates = 1;
+        }
+
+        /// <summary>
+        /// Frees the memory allocated for a node
+        /// </summary>
+        /// <param name="node">BTreeNode to free from memory</param>
+        private void Free(ref BTreeNode* node)
+        {
+            if (node == null)
+                return;
+
+            // If this is an internal node, free all its children first
+            if (node->info->type == BTreeNodeType.Internal)
+            {
+                for (int i = 0; i <= node->info->count; i++)
+                {
+                    var child = node->data.children[i];
+                    Free(ref child);
+                    node->data.children[i] = null;
+                }
+            }
+
+            // Free the memory handle
+            if (node->memoryHandle != null)
+            {
+                NativeMemory.Free(node->memoryHandle);
+                stats.numDeallocates++;
+                node = null;
+            }
+        }
+
+        /// <summary>
+        /// Frees the memory allocated for a node
+        /// </summary>
+        /// <param name="node"></param>
+        public static void FreeNode(ref BTreeNode* node)
+        {
+            if (node == null)
+                return;
+
+            // If this is an internal node, free all its children first
+            if (node->info->type == BTreeNodeType.Internal)
+            {
+                for (int i = 0; i <= node->info->count; i++)
+                {
+                    var child = node->data.children[i];
+                    FreeNode(ref child);
+                    node->data.children[i] = null;
+                }
+            }
+
+            // Free the memory handle
+            if (node->memoryHandle != null)
+            {
+                NativeMemory.Free(node->memoryHandle);
+                node = null;
+            }
+        }
+
+        /// <summary>
+        /// Deallocates the memory allocated for the B+Tree
+        /// </summary>
+        public void Deallocate()
+        {
+            if (root == null)
+                return;
+            Free(ref root);
+            Console.WriteLine("free complete");
+            stats.printStats();
+            root = null;
+            head = null;
+            tail = null;
+        }
+
+        /// <summary>
+        /// Destructor for the B+tree
+        /// </summary>
+        ~BTree()
+        {
+            Deallocate();
+        }
+
+        public ulong FastInserts => stats.totalFastInserts;
+        public ulong LeafCount => stats.numLeafNodes;
+        public ulong InternalCount => stats.numInternalNodes;
+
+        public ulong ValidCount => StatsValidCount();
+
+        public long RootValidCount => GetValidCount(root);
+
+        public long TailValidCount => GetValidCount(tail);
+
+        public long Count()
+        {
+            return stats.numKeys;
+        }
+        public ulong StatsValidCount()
+        {
+            return stats.numValidKeys;
+        }
+
+        public long GetValidCount(BTreeNode* node)
+        {
+            return node->info->validCount;
+        }
+
+        /// <summary>
+        /// Retrieves the first entry in the B+Tree (smallest key)
+        /// </summary>
+        /// <returns>entry fetched</returns>
+        public KeyValuePair<byte[], Value> First()
+        {
+            BTreeNode* leaf = head;
+            if (leaf == null)
+            {
+                return default;
+            }
+            byte[] keyBytes = new ReadOnlySpan<byte>(leaf->GetKey(0), BTreeNode.KEY_SIZE).ToArray();
+            return new KeyValuePair<byte[], Value>(keyBytes, leaf->GetValue(0));
+        }
+
+        /// <summary>
+        /// Retrieves the last entry in the B+Tree (largest key)
+        /// </summary>
+        /// <returns>entry fetched</returns>
+        public KeyValuePair<byte[], Value> Last()
+        {
+            BTreeNode* leaf = tail;
+            if (leaf == null)
+            {
+                return default;
+            }
+            byte[] keyBytes = new ReadOnlySpan<byte>(leaf->GetKey(leaf->info->count - 1), BTreeNode.KEY_SIZE).ToArray();
+            return new KeyValuePair<byte[], Value>(keyBytes, leaf->GetValue(leaf->info->count - 1));
+        }
+
+    }
+}

libs/server/BTreeIndex/BTreeDelete.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+namespace Garnet.server.BTreeIndex
+{
+    public unsafe partial class BTree
+    {
+        /// <summary>
+        /// Delete a key from the B+tree
+        /// </summary>
+        /// <param name="key">key to delete</param>
+        /// <returns>true if key was tombstoned</returns>
+        public bool Delete(byte* key)
+        {
+            BTreeNode* leaf = null;
+            var nodesTraversed = new BTreeNode*[MAX_TREE_DEPTH];
+
+            TraverseToLeaf(ref leaf, ref nodesTraversed, key);
+            var index = leaf->LowerBound(key);
+            if (index >= leaf->info->count || BTreeNode.Compare(key, leaf->GetKey(index)) != 0)
+            {
+                return false;
+            }
+
+            // insert a tombstone for the delete 
+            leaf->InsertTombstone(index);
+            leaf->info->validCount--;
+            stats.numValidKeys--;
+            return true;
+        }
+    }
+}