Step 13

db.c

@@ -124,7 +124,8 @@ const uint32_t INTERNAL_NODE_HEADER_SIZE = COMMON_NODE_HEADER_SIZE + INTERNAL_NO
 const uint32_t INTERNAL_NODE_KEY_SIZE = sizeof(uint32_t);
 const uint32_t INTERNAL_NODE_CHILD_SIZE = sizeof(uint32_t);
 const uint32_t INTERNAL_NODE_CELL_SIZE = INTERNAL_NODE_CHILD_SIZE + INTERNAL_NODE_KEY_SIZE;
-
+/* Keep this small for testing */
+const uint32_t INTERNAL_NODE_MAX_CELLS = 3;
 /*
  * Leaf Node Header Layout
  */
@@ -167,6 +168,10 @@ void set_node_root(void* node, bool is_root) {
     *((uint8_t*)(node + IS_ROOT_OFFSET)) = value;
 }
 
+uint32_t* node_parent(void* node) {
+    return node + PARENT_POINTER_OFFSET;
+}
+
 uint32_t* internal_node_num_keys(void* node) {
     return node + INTERNAL_NODE_NUM_KEYS_OFFSET;
 }
@@ -192,7 +197,7 @@ uint32_t* internal_node_child(void* node, uint32_t child_num) {
 }
 
 uint32_t* internal_node_key(void* node, uint32_t key_num) {
-    return internal_node_cell(node, key_num) + INTERNAL_NODE_CHILD_SIZE;
+    return (void*)internal_node_cell(node, key_num) + INTERNAL_NODE_CHILD_SIZE;
 }
 
 uint32_t* leaf_node_num_cells(void* node) {
@@ -359,11 +364,14 @@ Cursor* leaf_node_find(Table* table, uint32_t page_num, uint32_t key) {
     return cursor;
 }
 
-Cursor* internal_node_find(Table* table, uint32_t page_num, uint32_t key) {
-    void* node = get_page(table->pager, page_num);
+uint32_t internal_node_find_child(void* node, uint32_t key) {
+    /*
+     * Return the index of the child which should contain
+     * the given key
+     */
     uint32_t num_keys = *internal_node_num_keys(node);
 
-    /* Binary search to find index of child to search */
+    /* Binary search */
     uint32_t min_index = 0;
     uint32_t max_index = num_keys; /* there is one more child than key */
 
@@ -377,7 +385,14 @@ Cursor* internal_node_find(Table* table, uint32_t page_num, uint32_t key) {
         }
     }
 
-    uint32_t child_num = *internal_node_child(node, min_index);
+    return min_index;
+}
+
+Cursor* internal_node_find(Table* table, uint32_t page_num, uint32_t key) {
+    void *node = get_page(table->pager, page_num);
+
+    uint32_t child_index = internal_node_find_child(node, key);
+    uint32_t child_num = *internal_node_child(node, child_index);
     void* child = get_page(table->pager, child_num);
     switch (get_node_type(child)) {
         case NODE_LEAF:
@@ -654,6 +669,50 @@ void create_new_root(Table* table, uint32_t right_child_page_num) {
     uint32_t left_child_max_key = get_node_max_key(left_child);
     *internal_node_key(root, 0) = left_child_max_key;
     *internal_node_right_child(root) = right_child_page_num;
+    *node_parent(left_child) = table->root_page_num;
+    *node_parent(right_child) = table->root_page_num;
+}
+
+void internal_node_insert(Table* table, uint32_t parent_page_num, uint32_t child_page_num) {
+    /*
+     * Add a new child/key pair to parent that corresponds to child
+     */
+    void* parent = get_page(table->pager, parent_page_num);
+    void* child = get_page(table->pager, child_page_num);
+    uint32_t child_max_key = get_node_max_key(child);
+    uint32_t index = internal_node_find_child(parent, child_max_key);
+
+    uint32_t original_num_keys = *internal_node_num_keys(parent);
+    *internal_node_num_keys(parent) = original_num_keys + 1;
+
+    if (original_num_keys >= INTERNAL_NODE_MAX_CELLS) {
+        printf("Need to implement splitting internal node\n");
+        exit(EXIT_FAILURE);
+    }
+
+    uint32_t right_child_page_num = *internal_node_right_child(parent);
+    void* right_child = get_page(table->pager, right_child_page_num);
+
+    if (child_max_key > get_node_max_key(right_child)) {
+        /* Replace right child */
+        *internal_node_child(parent, original_num_keys) = right_child_page_num;
+        *internal_node_key(parent, original_num_keys) = get_node_max_key(right_child);
+        *internal_node_right_child(parent) = child_page_num;
+    } else {
+        /* Make room for new cell */
+        for (uint32_t i = original_num_keys; i > index; i--) {
+            void* destination = internal_node_cell(parent, i);
+            void* source = internal_node_cell(parent, i - 1);
+            memcpy(destination, source, INTERNAL_NODE_CELL_SIZE);
+        }
+        *internal_node_child(parent, index) = child_page_num;
+        *internal_node_key(parent, index) = child_max_key;
+    }
+}
+
+void update_internal_node_key(void* node, uint32_t old_key, uint32_t new_key) {
+    uint32_t old_child_index = internal_node_find_child(node, old_key);
+    *internal_node_key(node, old_child_index) = new_key;
 }
 
 void leaf_node_split_and_insert(Cursor* cursor, uint32_t key, Row* value) {
@@ -663,9 +722,11 @@ void leaf_node_split_and_insert(Cursor* cursor, uint32_t key, Row* value) {
      * Update parent or create a new parent.
      */
     void* old_node = get_page(cursor->table->pager, cursor->page_num);
+    uint32_t old_max = get_node_max_key(old_node);
     uint32_t new_page_num = get_unused_page_num(cursor->table->pager);
     void* new_node = get_page(cursor->table->pager, new_page_num);
     initialize_leaf_node(new_node);
+    *node_parent(new_node) = *node_parent(old_node);
     *leaf_node_next_leaf(new_node) = *leaf_node_next_leaf(old_node);
     *leaf_node_next_leaf(old_node) = new_page_num;
 
@@ -701,8 +762,13 @@ void leaf_node_split_and_insert(Cursor* cursor, uint32_t key, Row* value) {
     if (is_node_root(old_node)) {
         return create_new_root(cursor->table, new_page_num);
     } else {
-        printf("Need to implement updating parent after split\n");
-        exit(EXIT_FAILURE);
+       uint32_t parent_page_num = *node_parent(old_node);
+       uint32_t new_max = get_node_max_key(old_node);
+       void* parent = get_page(cursor->table->pager, parent_page_num);
+
+       update_internal_node_key(parent, old_max, new_max);
+       internal_node_insert(cursor->table, parent_page_num, new_page_num);
+       return;
     }
 }