devbox
AllinToyou
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

构建一个简单的数据库系列（六）cursor抽象_用cursor.so完整开发一下项目

作者：AllinToyou | 2024-03-27 12:55:42

赞

踩

用cursor.so完整开发一下项目

英文连接：https://cstack.github.io/db_tutorial/parts/part6.html

为了实现B-tree，这一节先对当前的实现进行一点重构。

我们增加一个概念：Cursor（游标），代表了对象在数据库中的位置。那么关于cursor有几件事需要完成

1、在表之前创建cursor

2、在表之后创建cursor

3、访问cursor所指向的行

4、访问cursor的下一行

完成这些之后，我们还会继续实现：

1、使用cursor删除行

2、使用cursor修改行

3、使用ID查询表，并且在在该行后创建cursor

在没有实现ado，Cursor类型暂时按如下定义：

  1. 批注：ADO 连接对象（ADO Connection Object）
  2. ADO 连接对象用来创建到某个数据源的开放连接。通过此连接，您可以对此数据库进行访问和操作。
  3. 查看此连接对象的所有方法和属性
  1. struct Cursor_t {
  2.   Table* table;
  3.   uint32_t row_num;
  4.   bool end_of_table;  // Indicates a position one past the last element
  5. };
  6. typedef struct Cursor_t Cursor;

table_start() and table_end() create new cursors:

  1. Cursor* table_start(Table* table) {
  2.       Cursor* cursor = malloc(sizeof(Cursor));
  3.       cursor->table = table;
  4.       cursor->row_num = 0;
  5.       cursor->end_of_table = (table->num_rows == 0);
  6.     
  7.               return cursor;
  8.     }
  9.  
  10. Cursor* table_end(Table* table) {
  11.       Cursor* cursor = malloc(sizeof(Cursor));
  12.       cursor->table = table;
  13.       cursor->row_num = table->num_rows;
  14.       cursor->end_of_table = true;
  15.     
  16.               return cursor;
  17.     }

row_slot（）修改为cursor_value(), 该函数作用：指向了sursor所执行的位置

定义1个函数cursor_advance，实现对num_rows加1。

  1. void cursor_advance(Cursor* cursor) {
  2.       cursor->row_num = 1;
  3.       if (cursor->row_num >= cursor->table->num_rows) {
  4.             cursor->end_of_table = true;
  5.           }
  6.     }

最后，我们修改“virtual machine”，改用抽象的对象：Cursor。当插入一行时，我们打开一个Cursor，指向表尾。在cursor后写入后，关闭Cursor。

  1.    Row* row_to_insert = &(statement->row_to_insert);
  2. +  Cursor* cursor = table_end(table);
  3.  
  4. -  serialize_row(row_to_insert, row_slot(table, table->num_rows));
  5. +  serialize_row(row_to_insert, cursor_value(cursor));
  6.    table->num_rows += 1;
  7.  
  8. +  free(cursor);
  9. +
  10.    return EXECUTE_SUCCESS;
  11.  }

同理，修改execute_select的实现，使用cursor替换row_slot

  1. ExecuteResult execute_select(Statement* statement, Table* table) {
  2. +  Cursor* cursor = table_start(table);
  3. +
  4.    Row row;
  5. -  for (uint32_t i = 0; i < table->num_rows; i++) {
  6. -    deserialize_row(row_slot(table, i), &row);
  7. +  while (!(cursor->end_of_table)) {
  8. +    deserialize_row(cursor_value(cursor), &row);
  9.      print_row(&row);
  10. +    cursor_advance(cursor);
  11.    }
  12. +
  13. +  free(cursor);
  14. +
  15.    return EXECUTE_SUCCESS;
  16.  }

至此，execute_select和execute_insert就不需要在做任何关于表存储的假设，就可以通过cursor和table进行交互了。

还是使用上节的用例测试下：

  1. db > insert 1 cstack foo@bar.com
  2. Executed.
  3. db > insert 2 hello hello@126.com
  4. Executed.
  5. db > select
  6. (1, cstack, foo@bar.com)
  7. (2, hello, hello@126.com)
  8. (1, cstack, foo@bar.com)
  9. (2, hello, hello@126.com)
  10. Executed.
  11. db >

至此最新代码：

  1. #include <errno.h>
  2. #include <fcntl.h>
  3. #include <stdbool.h>
  4. #include <stdio.h>
  5. #include <stdlib.h>
  6. #include <string.h>
  7. #include <unistd.h>
  8.  
  9. /* 定义 元数据操作结果*/
  10. enum MetaCommandResult_t {
  11.     META_COMMAND_SUCCESS,
  12.     META_COMMAND_UNRECOGNIZED_COMMAND
  13. };
  14. typedef enum MetaCommandResult_t MetaCommandResult;
  15.  
  16. /* 执行结果*/
  17. enum ExecuteResult_t { EXECUTE_SUCCESS, EXECUTE_TABLE_FULL };
  18. typedef enum ExecuteResult_t ExecuteResult;
  19.  
  20. /* sql解析结果 */
  21. enum PrepareResult_t {
  22.     PREPARE_SUCCESS,
  23.     PREPARE_NEGATIVE_ID,
  24.     PREPARE_STRING_TOO_LONG,
  25.     PREPARE_SYNTAX_ERROR,
  26.     PREPARE_UNRECOGNIZED_STATEMENT
  27. };
  28. typedef enum PrepareResult_t PrepareResult;
  29.  
  30. /* 行定义，对应具体的业务 */
  31. const uint32_t COLUMN_USERNAME_SIZE = 32;
  32. const uint32_t COLUMN_EMAIL_SIZE = 255;
  33. struct Row_t {
  34.       uint32_t id;
  35.       char username[COLUMN_USERNAME_SIZE + 1];
  36.       char email[COLUMN_EMAIL_SIZE + 1];
  37. };
  38. typedef struct Row_t Row;
  39.  
  40.  
  41. /* sql type*/
  42. enum StatementType_t{
  43.     STATEMENT_INSERT,
  44.     STATEMENT_SELECT
  45. };
  46.  
  47. typedef enum StatementType_t StatementType;
  48.  
  49. struct Statement_t {
  50.     StatementType type;
  51.     Row row_to_insert;  /* only used by insert statement */
  52.  
  53. };
  54. typedef struct Statement_t Statement;
  55.  
  56. /**/
  57.  
  58. #define size_of_attribute(Struct, Attribute) sizeof(((Struct*)0)->Attribute)
  59.  
  60. const uint32_t ID_SIZE = size_of_attribute(Row, id);
  61. const uint32_t USERNAME_SIZE = size_of_attribute(Row, username);
  62. const uint32_t EMAIL_SIZE = size_of_attribute(Row, email);
  63. const uint32_t ID_OFFSET = 0;
  64. const uint32_t USERNAME_OFFSET = ID_OFFSET + ID_SIZE;
  65. const uint32_t EMAIL_OFFSET = USERNAME_OFFSET + USERNAME_SIZE;
  66. const uint32_t ROW_SIZE = ID_SIZE + USERNAME_SIZE + EMAIL_SIZE;
  67.  
  68. /* 大部分系统结构的页大小都是4K ，所以这里也定义为4K 这样就不用做转换 */
  69. const uint32_t PAGE_SIZE = 4096;
  70. const uint32_t TABLE_MAX_PAGES = 100;
  71. const uint32_t ROWS_PER_PAGE = PAGE_SIZE / ROW_SIZE;
  72. const uint32_t TABLE_MAX_ROWS = ROWS_PER_PAGE * TABLE_MAX_PAGES;
  73.  
  74. /*
  75.  * 页和表定义
  76.  */
  77. struct Pager_t {
  78.     int file_descriptor;
  79.     uint32_t file_length;
  80.     void *pages[TABLE_MAX_PAGES];
  81. };
  82. typedef struct Pager_t Pager;
  83.  
  84. struct Table_t {
  85.     Pager *pager;
  86.     uint32_t num_rows;
  87. };
  88. typedef struct Table_t Table;
  89.  
  90. /* 序列化： 将row写到内存中 */
  91. void serialize_row(Row* source, void* destination) {
  92.     memcpy(destination + ID_OFFSET, &(source->id), ID_SIZE);
  93.     memcpy(destination + USERNAME_OFFSET, &(source->username), USERNAME_SIZE);
  94.     memcpy(destination + EMAIL_OFFSET, &(source->email), EMAIL_SIZE);
  95. }
  96.  
  97. /* 反序列化 */
  98. void deserialize_row(void* source, Row* destination) {
  99.     memcpy(&(destination->id), source + ID_OFFSET, ID_SIZE);
  100.     memcpy(&(destination->username), source + USERNAME_OFFSET, USERNAME_SIZE);
  101.     memcpy(&(destination->email), source + EMAIL_OFFSET, EMAIL_SIZE);
  102. }
  103.  
  104. /**
  105.  * 获取页num对应的页
  106.  * @param table
  107.  * @param row_num
  108.  * @return
  109.  */
  110. void *get_page(Pager *pager, uint32_t page_num)
  111. {
  112.     if (page_num > TABLE_MAX_PAGES) {
  113.         printf("Tried to fetch page number out of bounds. %d > %d\n", page_num,
  114.                TABLE_MAX_PAGES);
  115.         exit(EXIT_FAILURE);
  116.     }
  117.  
  118.     if (pager->pages[page_num] == NULL) {
  119.         // Cache miss. Allocate memory and load from file.
  120.         void *page = malloc(PAGE_SIZE);
  121.         uint32_t num_pages = pager->file_length / PAGE_SIZE;
  122.  
  123.         // We might save a partial page at the end of the file
  124.         if (pager->file_length % PAGE_SIZE) {
  125.             num_pages += 1;
  126.         }
  127.  
  128.         if (page_num <= num_pages) {
  129.             lseek(pager->file_descriptor, page_num * PAGE_SIZE, SEEK_SET);
  130.             ssize_t bytes_read = read(pager->file_descriptor, page, PAGE_SIZE);
  131.             if (bytes_read == -1) {
  132.                 printf("Error reading file: %d\n", errno);
  133.                 exit(EXIT_FAILURE);
  134.             }
  135.         }
  136.  
  137.         pager->pages[page_num] = page;
  138.     }
  139.  
  140.     return pager->pages[page_num];
  141. }
  142.  
  143.  
  144. /**
  145.  * 打开pager
  146.  * @param filename
  147.  * @return
  148.  */
  149. Pager *pager_open(const char *filename)
  150. {
  151.     int fd = open(filename,
  152.                   O_RDWR |      // Read/Write mode
  153.                   O_CREAT,  // Create file if it does not exist
  154.                   S_IWUSR |     // User write permission
  155.                   S_IRUSR   // User read permission
  156.     );
  157.  
  158.     if (fd == -1) {
  159.         printf("Unable to open file\n");
  160.         exit(EXIT_FAILURE);
  161.     }
  162.  
  163.     off_t file_length = lseek(fd, 0, SEEK_END);
  164.  
  165.     Pager *pager = malloc(sizeof(Pager));
  166.     pager->file_descriptor = fd;
  167.     pager->file_length = file_length;
  168.  
  169.     for (uint32_t i = 0; i < TABLE_MAX_PAGES; i++) {
  170.         pager->pages[i] = NULL;
  171.     }
  172.  
  173.     return pager;
  174. }
  175.  
  176. /**
  177.  * Cursor定义
  178.  */
  179. struct Cursor_t {
  180.     Table* table;
  181.     uint32_t row_num;
  182.     bool end_of_table;  // Indicates a position one past the last element
  183. };
  184. typedef struct Cursor_t Cursor;
  185.  
  186. /**
  187.  * Cursor api
  188.  * @param row
  189.  */
  190. Cursor* table_start(Table* table) {
  191.     Cursor* cursor = malloc(sizeof(Cursor));
  192.     cursor->table = table;
  193.     cursor->row_num = 0;
  194.     cursor->end_of_table = (table->num_rows == 0);
  195.  
  196.     return cursor;
  197. }
  198.  
  199. Cursor* table_end(Table* table) {
  200.     Cursor* cursor = malloc(sizeof(Cursor));
  201.     cursor->table = table;
  202.     cursor->row_num = table->num_rows;
  203.     cursor->end_of_table = true;
  204.  
  205.     return cursor;
  206. }
  207.  
  208.  
  209. /**
  210.  * 计算插入位置， 行插入槽
  211.  */
  212. void* cursor_value(Cursor* cursor)
  213. {
  214.     uint32_t row_num = cursor->row_num;
  215.     uint32_t page_num = row_num / ROWS_PER_PAGE;
  216.     void* page = get_page(cursor->table->pager, page_num);
  217.     uint32_t row_offset = row_num % ROWS_PER_PAGE;
  218.     uint32_t byte_offset = row_offset * ROW_SIZE;
  219.     return page + byte_offset;
  220. }
  221.  
  222. /**
  223.  * rownum 加1
  224.  * @param cursor
  225.  */
  226. void cursor_advance(Cursor* cursor)
  227. {
  228.     cursor->row_num += 1;
  229.     if (cursor->row_num >= cursor->table->num_rows) {
  230.         cursor->end_of_table = true;
  231.     }
  232. }
  233.  
  234.  
  235.  
  236. void print_row(Row* row) {
  237.     printf("(%d, %s, %s)\n", row->id, row->username, row->email);
  238. }
  239.  
  240. /**
  241.  * 打开数据库文件并建立连接
  242.  * @param filename
  243.  * @return
  244.  */
  245. Table *db_open(const char *filename)
  246. {
  247.     Pager *pager = pager_open(filename);
  248.     uint32_t num_rows = pager->file_length / ROW_SIZE;
  249.  
  250.     Table *table = malloc(sizeof(Table));
  251.     table->num_rows = 0;
  252.     table->pager = pager;
  253.     table->num_rows = num_rows;
  254.  
  255.     return table;
  256. }
  257.  
  258. /* 接收输入*/
  259. struct InputBuffer_t {
  260.     char* buffer;
  261.     size_t buffer_length;
  262.     ssize_t input_length;
  263. };
  264. typedef struct InputBuffer_t InputBuffer;
  265.  
  266. /* 初始化buffer */
  267. InputBuffer* new_input_buffer()
  268. {
  269.     InputBuffer* input_buffer = malloc(sizeof(InputBuffer));
  270.     input_buffer->buffer = NULL;
  271.     input_buffer->buffer_length = 0;
  272.     input_buffer->input_length = 0;
  273.  
  274.     return input_buffer;
  275. }
  276.  
  277. void print_prompt() { printf("db > "); }
  278.  
  279. /* 按行从标准输入读取 */
  280. void read_input(InputBuffer* input_buffer)
  281. {
  282.     ssize_t bytes_read =
  283.             getline(&(input_buffer->buffer), &(input_buffer->buffer_length), stdin);
  284.  
  285.     if (bytes_read <= 0) {
  286.         printf("Error reading input\n");
  287.         exit(EXIT_FAILURE);
  288.     }
  289.  
  290.     // Ignore trailing newline
  291.     input_buffer->input_length = bytes_read - 1;
  292.     input_buffer->buffer[bytes_read - 1] = 0;
  293. }
  294.  
  295. /**
  296.  * 刷新页到文件（指定页号）
  297.  * @param pager
  298.  * @param page_num
  299.  * @param size
  300.  */
  301. void pager_flush(Pager *pager, uint32_t page_num, uint32_t size)
  302. {
  303.     if (pager->pages[page_num] == NULL) {
  304.         printf("Tried to flush null page\n");
  305.         exit(EXIT_FAILURE);
  306.     }
  307.  
  308.     off_t offset = lseek(pager->file_descriptor, page_num * PAGE_SIZE, SEEK_SET);
  309.  
  310.     if (offset == -1) {
  311.         printf("Error seeking: %d\n", errno);
  312.         exit(EXIT_FAILURE);
  313.     }
  314.  
  315.     ssize_t bytes_written =
  316.             write(pager->file_descriptor, pager->pages[page_num], size);
  317.  
  318.     if (bytes_written == -1) {
  319.         printf("Error writing: %d\n", errno);
  320.         exit(EXIT_FAILURE);
  321.     }
  322. }
  323.  
  324. /**
  325.  * 关闭数据连接
  326.  */
  327. void db_close(Table *table) {
  328.     Pager *pager = table->pager;
  329.     uint32_t num_full_pages = table->num_rows / ROWS_PER_PAGE;
  330.  
  331.     for (uint32_t i = 0; i < num_full_pages; i++) {
  332.         if (pager->pages[i] == NULL) {
  333.             continue;
  334.         }
  335.         pager_flush(pager, i, PAGE_SIZE);
  336.         free(pager->pages[i]);
  337.         pager->pages[i] = NULL;
  338.     }
  339.  
  340.     // There may be a partial page to write to the end of the file
  341.     // This should not be needed after we switch to a B-tree
  342.     uint32_t num_additional_rows = table->num_rows % ROWS_PER_PAGE;
  343.     if (num_additional_rows > 0) {
  344.         uint32_t page_num = num_full_pages;
  345.         if (pager->pages[page_num] != NULL) {
  346.             pager_flush(pager, page_num, num_additional_rows * ROW_SIZE);
  347.             free(pager->pages[page_num]);
  348.             pager->pages[page_num] = NULL;
  349.         }
  350.     }
  351.  
  352.     int result = close(pager->file_descriptor);
  353.     if (result == -1) {
  354.         printf("Error closing db file.\n");
  355.         exit(EXIT_FAILURE);
  356.     }
  357.     for (uint32_t i = 0; i < TABLE_MAX_PAGES; i++) {
  358.         void *page = pager->pages[i];
  359.         if (page) {
  360.             free(page);
  361.             pager->pages[i] = NULL;
  362.         }
  363.     }
  364.     free(pager);
  365. }
  366.  
  367.  
  368. /* 元数据命令处理 */
  369. MetaCommandResult do_meta_command(InputBuffer* input_buffer, Table* table)
  370. {
  371.     if (strcmp(input_buffer->buffer, ".exit") == 0) {
  372.     db_close(table);
  373.         exit(EXIT_SUCCESS);
  374.     } else {
  375.         return META_COMMAND_UNRECOGNIZED_COMMAND;
  376.     }
  377. }
  378.  
  379. /* insert解析和校验*/
  380. PrepareResult prepare_insert(InputBuffer *input_buffer, Statement *statement) {
  381.     statement->type = STATEMENT_INSERT;
  382.  
  383.     char *keyword = strtok(input_buffer->buffer, " ");
  384.     char *id_string = strtok(NULL, " ");
  385.     char *username = strtok(NULL, " ");
  386.     char *email = strtok(NULL, " ");
  387.  
  388.     if (id_string == NULL || username == NULL || email == NULL) {
  389.         return PREPARE_SYNTAX_ERROR;
  390.     }
  391.  
  392.     int id = atoi(id_string);
  393.     if (id < 0) {
  394.         return PREPARE_NEGATIVE_ID;
  395.     }
  396.     if (strlen(username) > COLUMN_USERNAME_SIZE) {
  397.         return PREPARE_STRING_TOO_LONG;
  398.     }
  399.     if (strlen(email) > COLUMN_EMAIL_SIZE) {
  400.         return PREPARE_STRING_TOO_LONG;
  401.     }
  402.  
  403.     statement->row_to_insert.id = id;
  404.     strcpy(statement->row_to_insert.username, username);
  405.     strcpy(statement->row_to_insert.email, email);
  406.  
  407.     return PREPARE_SUCCESS;
  408. }
  409.  
  410. /* sql解析 */
  411. PrepareResult prepare_statement(InputBuffer* input_buffer,Statement* statement)
  412. {
  413.     if (strncasecmp(input_buffer->buffer, "insert", 6) == 0) {
  414.         return prepare_insert(input_buffer, statement);
  415.     }
  416.     if (strncasecmp(input_buffer->buffer, "select", 6) == 0) {
  417.         statement->type = STATEMENT_SELECT;
  418.         return PREPARE_SUCCESS;
  419.     }
  420.  
  421.     return PREPARE_UNRECOGNIZED_STATEMENT;
  422. }
  423.  
  424. /* 执行insert*/
  425. ExecuteResult execute_insert(Statement *statement, Table *table)
  426. {
  427.     if (table->num_rows >= TABLE_MAX_ROWS) {
  428.         return EXECUTE_TABLE_FULL;
  429.     }
  430.  
  431.     Row *row_to_insert = &(statement->row_to_insert);
  432.     Cursor* cursor = table_end(table);
  433.  
  434.     serialize_row(row_to_insert, cursor_value(cursor));
  435.     table->num_rows += 1;
  436.  
  437.     free(cursor);
  438.  
  439.     return EXECUTE_SUCCESS;
  440. }
  441.  
  442. /* 执行查询*/
  443. ExecuteResult execute_select(Statement *statement, Table *table)
  444. {
  445.     Row row;
  446.     Cursor* cursor = table_start(table);
  447.  
  448.     while (!(cursor->end_of_table)) {
  449.         deserialize_row(cursor_value(cursor), &row);
  450.         print_row(&row);
  451.         cursor_advance(cursor);
  452.     }
  453.  
  454.     free(cursor);
  455.  
  456.     return EXECUTE_SUCCESS;
  457. }
  458. /* sql执行*/
  459. ExecuteResult execute_statement(Statement* statement , Table* table)
  460. {
  461.     switch (statement->type)
  462.     {
  463.         case (STATEMENT_INSERT):
  464.             return execute_insert(statement, table);
  465.         case (STATEMENT_SELECT):
  466.             return execute_select(statement, table);
  467.     }
  468. }
  469.  
  470.  
  471.  
  472. /* 主函数*/
  473. int main(int argc, char* argv[])
  474. {
  475.     if (argc < 2) {
  476.         printf("Must supply a database filename.\n");
  477.         exit(EXIT_FAILURE);
  478.     }
  479.  
  480.     char *filename = argv[1];
  481.     Table *table = db_open(filename);
  482.  
  483.     InputBuffer *input_buffer = new_input_buffer();
  484.     while (true)
  485.     {
  486.         print_prompt();
  487.         read_input(input_buffer);
  488.  
  489.         if (input_buffer->buffer[0] == '.')
  490.         {
  491.             switch (do_meta_command(input_buffer,table))
  492.             {
  493.                 case (META_COMMAND_SUCCESS):
  494.                     continue;
  495.                 case (META_COMMAND_UNRECOGNIZED_COMMAND):
  496.                     printf("Unrecognized command '%s'\n", input_buffer->buffer);
  497.                     continue;
  498.             }
  499.         }
  500.  
  501.         Statement statement;
  502.         switch (prepare_statement(input_buffer, &statement))
  503.         {
  504.             case (PREPARE_SUCCESS):
  505.                 break;
  506.             case (PREPARE_NEGATIVE_ID):
  507.                 printf("ID must be positive.\n");
  508.                 continue;
  509.             case (PREPARE_STRING_TOO_LONG):
  510.                 printf("String is too long.\n");
  511.                 continue;
  512.             case (PREPARE_SYNTAX_ERROR):
  513.                 printf("Syntax error. Could not parse statement.\n");
  514.                 continue;
  515.             case (PREPARE_UNRECOGNIZED_STATEMENT):
  516.                 printf("Unrecognized keyword at start of '%s'.\n",
  517.                        input_buffer->buffer);
  518.                 continue;
  519.         }
  520.  
  521.         switch (execute_statement(&statement, table))
  522.         {
  523.             case (EXECUTE_SUCCESS):
  524.                 printf("Executed.\n");
  525.                 break;
  526.             case (EXECUTE_TABLE_FULL):
  527.                 printf("Error: Table full.\n");
  528.                 break;
  529.         }
  530.     }
  531. }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/AllinToyou/article/detail/324274
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号