Merge pull request #8 from dasatomic/custered_index

Clustered index

DataStructures/BTreeCollection.cs

@@ -76,6 +76,54 @@ public BTreeCollection(IAllocateMixedPage pageAllocator, ColumnInfo[] columnType
             this.projectionRemoveIndexCol = new ProjectExtendInfo(mps, prjs, new ColumnInfo[0]);
         }
 
+        public BTreeCollection(IAllocateMixedPage pageAllocator, ColumnInfo[] columnTypes, Func<RowHolder, RowHolder, int> indexComparer, int indexPosition, ulong initialPageId, ITransaction tran)
+        {
+            if (pageAllocator == null || columnTypes == null || columnTypes.Length == 0 || tran == null)
+            {
+                throw new ArgumentNullException();
+            }
+
+            this.pageAllocator = pageAllocator;
+            this.columnTypes = columnTypes;
+            this.indexPosition = indexPosition;
+
+            // Add PagePointer to the end of each row.
+            // Page pointer will be used as pointer in btree.
+            // Prev page pointer will point to the left side of the first key.
+            // Next page pointer is used to encode additional info.
+            // For now next page is used to encode whether the page is leaf.
+            this.btreeColumnTypes = new ColumnInfo[columnTypes.Length + 1];
+            Array.Copy(columnTypes, this.btreeColumnTypes, columnTypes.Length);
+            this.btreeColumnTypes[columnTypes.Length] = new ColumnInfo(ColumnType.PagePointer);
+            this.collectionRootPageId = initialPageId;
+
+            using Releaser lck = tran.AcquireLock(initialPageId, LockManager.LockTypeEnum.Shared).ConfigureAwait(false).GetAwaiter().GetResult();
+            MixedPage rootPage = pageAllocator.GetMixedPage(this.collectionRootPageId, tran, columnTypes).Result;
+            uint maxRowCount = rootPage.MaxRowCount();
+
+            // In btree we can fit max 2*t + 1 elements.
+            // If we can fit even number we will keep the space unused.
+            this.maxElemsPerPage = (int)maxRowCount;
+            if (maxRowCount % 2 == 0)
+            {
+                this.maxElemsPerPage--;
+            }
+
+            this.indexComparer = indexComparer;
+
+            this.pagePointerRowPosition = this.columnTypes.Length;
+
+            MappingType[] mps = new MappingType[columnTypes.Length];
+            int[] prjs = new int[columnTypes.Length];
+            for (int i = 0; i < prjs.Length; i++)
+            {
+                mps[i] = MappingType.Projection;
+                prjs[i] = i;
+            }
+
+            this.projectionRemoveIndexCol = new ProjectExtendInfo(mps, prjs, new ColumnInfo[0]);
+        }
+
         public async Task Add(RowHolder item, ITransaction tran)
         {
             // Extend RowHolder with page pointer.

DataStructures/PageListCollection.cs

@@ -6,6 +6,39 @@
 
 namespace DataStructures
 {
+    public enum CollectionType
+    {
+        PageList,
+        BTree,
+    }
+
+    public static class CollectionHandler
+    {
+        public static async Task<MixedPage> CreateInitialPage(CollectionType collectionType, ColumnInfo[] columnInfos, IAllocateMixedPage pageAllocator, ITransaction tran)
+        {
+            ArgumentNullException.ThrowIfNull(collectionType);
+            ArgumentNullException.ThrowIfNull(pageAllocator);
+            ArgumentNullException.ThrowIfNull(tran);
+
+            if (collectionType == CollectionType.PageList)
+            {
+                return await pageAllocator.AllocateMixedPage(columnInfos, PageManagerConstants.NullPageId, PageManagerConstants.NullPageId, tran);
+            }
+            else if (collectionType == CollectionType.BTree)
+            {
+                // Need to extend it to add page pointers.
+                ColumnInfo[] btreeColumnTypes = new ColumnInfo[columnInfos.Length + 1];
+                Array.Copy(columnInfos, btreeColumnTypes, columnInfos.Length);
+                btreeColumnTypes[columnInfos.Length] = new ColumnInfo(ColumnType.PagePointer);
+                return await pageAllocator.AllocateMixedPage(btreeColumnTypes, PageManagerConstants.NullPageId, PageManagerConstants.NullPageId, tran);
+            }
+            else
+            {
+                throw new ArgumentException();
+            }
+        }
+    }
+
     public interface IPageCollection<T>
     {
         Task<ulong> Count(ITransaction tran);

MetadataManager/Exceptions/OnlyOneClusteredIndexSupportedException.cs

@@ -0,0 +1,8 @@
+using System;
+
+namespace MetadataManager.Exceptions
+{
+    public class OnlyOneClusteredIndexSupportedException : Exception
+    {
+    }
+}

MetadataManager/MetadataColumns.cs

@@ -8,27 +8,32 @@ namespace MetadataManager
 {
     public struct MetadataColumn
     {
+        public const int NotPartOfClusteredIndex = -1;
+
         public const int ColumnIdColumnPos = 0;
         public readonly int ColumnId;
         public const int TableIdColumnPos = 1;
         public readonly int TableId;
         public const int ColumnNameColumnPos = 2;
         public readonly string ColumnName;
         public const int ColumnTypeColumnPos = 3;
-        public const int ColumnTypeLength = 4;
         public readonly ColumnInfo ColumnType;
+        public const int ColumnTypeLength = 4;
+        public const int ClusteredIndexPartPos = 5;
+        public readonly int ClusteredIndexPart;
 
-        public MetadataColumn(int columnId, int tableId, string columnName, ColumnInfo columnInfo)
+        public MetadataColumn(int columnId, int tableId, string columnName, ColumnInfo columnInfo, int clusteredIndexPart = NotPartOfClusteredIndex)
         {
             this.ColumnId = columnId;
             this.TableId = tableId;
             this.ColumnName = columnName;
             this.ColumnType = columnInfo;
+            this.ClusteredIndexPart = clusteredIndexPart;
         }
 
         public MetadataColumn SetNewColumnPosition(int columnId)
         {
-            return new MetadataColumn(columnId, this.TableId, this.ColumnName, this.ColumnType);
+            return new MetadataColumn(columnId, this.TableId, this.ColumnName, this.ColumnType, this.ClusteredIndexPart);
         }
     }
 
@@ -38,12 +43,13 @@ public MetadataColumn SetNewColumnPosition(int columnId)
     /// </summary>
     public struct ColumnCreateDefinition
     {
-        public ColumnCreateDefinition(int tableId, string columnName, ColumnInfo columnInfo, int columnId)
+        public ColumnCreateDefinition(int tableId, string columnName, ColumnInfo columnInfo, int columnId, int clusteredIndexPosition)
         {
             this.TableId = tableId;
             this.ColumnName = columnName;
             this.ColumnType = columnInfo;
             this.ColumnId = columnId;
+            this.ClusteredIndexPosition = clusteredIndexPosition;
         }
 
         /// <summary>
@@ -65,6 +71,12 @@ public ColumnCreateDefinition(int tableId, string columnName, ColumnInfo columnI
         /// Column position in table.
         /// </summary>
         public int ColumnId { get; }
+
+        /// <summary>
+        /// Position in clustered index, if this column is part of it.
+        /// -1 otherwise.
+        /// </summary>
+        public int ClusteredIndexPosition { get; }
     }
 
     public class MetadataColumnsManager : IMetadataObjectManager<MetadataColumn, ColumnCreateDefinition, Tuple<int, int> /* column id - table id */>
@@ -83,6 +95,7 @@ public class MetadataColumnsManager : IMetadataObjectManager<MetadataColumn, Col
             new ColumnInfo(ColumnType.String, MAX_NAME_LENGTH), // pointer to name
             new ColumnInfo(ColumnType.Int), // column type
             new ColumnInfo(ColumnType.Int), // column type length
+            new ColumnInfo(ColumnType.Int), // clustered index position.
         };
 
         public static ColumnInfo[] GetSchemaDefinition() => columnDefinitions;
@@ -109,7 +122,8 @@ await foreach (RowHolder rh in pageListCollection.Iterate(tran))
                     rh.GetField<int>(MetadataColumn.ColumnIdColumnPos),
                         rh.GetField<int>(MetadataColumn.TableIdColumnPos),
                         new string(columnName),
-                        new ColumnInfo((ColumnType)rh.GetField<int>(MetadataColumn.ColumnTypeColumnPos), rh.GetField<int>(MetadataColumn.ColumnTypeLength)));
+                        new ColumnInfo((ColumnType)rh.GetField<int>(MetadataColumn.ColumnTypeColumnPos), rh.GetField<int>(MetadataColumn.ColumnTypeLength)),
+                        rh.GetField<int>(MetadataColumn.ClusteredIndexPartPos));
             }
         }
 
@@ -123,11 +137,12 @@ public async Task<int> CreateObject(ColumnCreateDefinition def, ITransaction tra
             RowHolder rh = new RowHolder(columnDefinitions);
             PagePointerOffsetPair namePointer =  await this.stringHeap.Add(def.ColumnName.ToCharArray(), tran);
 
-            rh.SetField<int>(0, def.ColumnId);
-            rh.SetField<int>(1, def.TableId);
-            rh.SetField<PagePointerOffsetPair>(2, namePointer);
-            rh.SetField<int>(3, (int)def.ColumnType.ColumnType);
-            rh.SetField<int>(4, def.ColumnType.RepCount);
+            rh.SetField<int>(MetadataColumn.ColumnIdColumnPos, def.ColumnId);
+            rh.SetField<int>(MetadataColumn.TableIdColumnPos, def.TableId);
+            rh.SetField<PagePointerOffsetPair>(MetadataColumn.ColumnNameColumnPos, namePointer);
+            rh.SetField<int>(MetadataColumn.ColumnTypeColumnPos, (int)def.ColumnType.ColumnType);
+            rh.SetField<int>(MetadataColumn.ColumnTypeLength, def.ColumnType.RepCount);
+            rh.SetField<int>(MetadataColumn.ClusteredIndexPartPos, def.ClusteredIndexPosition);
 
             await pageListCollection.Add(rh, tran);
 
@@ -144,7 +159,8 @@ await foreach (RowHolder rh in pageListCollection.Iterate(tran))
                 {
                     PagePointerOffsetPair stringPointer = rh.GetField<PagePointerOffsetPair>(MetadataColumn.ColumnNameColumnPos);
 
-                    if (def.ColumnName == new string(await stringHeap.Fetch(stringPointer, tran)))
+                    char[] strContent = await stringHeap.Fetch(stringPointer, tran);
+                    if (CharrArray.Compare(strContent, def.ColumnName) == 0)
                     {
                         return true;
                     }