finish insert row & find row by pk

api/table.md

@@ -31,6 +31,8 @@
 > 区块长度固定，但是不同索引类型有所不同
 详见[索引格式](/api/index.md)。
 ### 表项
+> 实现时为了性能考虑，建表时限制数据栏数小于`128`
+
 > 区块长度固定，为`8+len(row 1)+len(row 2)+...+len(row N)`字节
 
 > 由于存在可变长的 string 和 binary，不同区块长度有可能不同

api/types.md

@@ -15,27 +15,34 @@
 | 5 | double | 直接存储 |
 
 ## 字符串
-> 字符串为变长的可索引的字节数组，长度最大不超过uint16(65535字节)。
+> 字符串为变长的可索引的字节数组，长度最大不超过uint16(65535字节)。实际上出于性能考虑，实现时限制长度不超过`2048`字节。
+
+> 一旦字符串插入表中，其长度就不再可变。如需改变需要删除再添加。
 ### 类型列表
 | 类型代号 | 类型 | 存储方式 |
 | --- | --- | --- |
 | 6 | string | 下一个哈希相同的数据项的指针+uint16长度+该长度的数据(字节数组) |
 
 ## 二进制数据块
-> 二进制数据块为变长的不参与索引的字节数组，长度最大不超过uint32。
+> 二进制数据块为变长的不参与索引的字节数组，长度最大不超过uint16(65535字节)。实际上出于性能考虑，实现时限制长度不超过`2048`字节。
+
+> 一旦二进制数据块插入表中，其长度就不再可变。如需改变需要删除再添加。
 ### 类型列表
 | 类型代号 | 类型 | 存储方式 |
 | --- | --- | --- |
-| 7 | binary | uint32长度+直接存储的数据(字节数组) |
+| 7 | binary | uint16长度+直接存储的数据(字节数组) |
 
 ## 类型修饰符
 > 类型修饰符占据类型代号的高2位，通过或运算与基础代号结合
 
+> 对于可空属性，会在其值前添加一字节，指示其是否为空，如果为空则为 true
+
+> 对于外键修饰符，将会在表头标明其外键连接到的表头指针
+
 | 类型代号 | 类型 | 说明 |
 | --- | --- | --- |
 | 0x00 | null | 无修饰：允许重复、可空、非外键 |
-| 0x40 | unique | 不允许重复 |
-| 0x80 | nonnull | 非空 |
-| 0xc0 | foreignkey | 外键 |
+| 0x10 | unique | 不允许重复 |
+| 0x20 | nonnull | 非空 |
+| 0x40 | foreignkey | 外键 |
 
-特别地，对于外键修饰符，将会在表头标明其外键连接到的表头指针。

include/table.h

@@ -4,7 +4,8 @@
 #include <stdint.h>
 #include "types.h"
 
-// 创建表，可变参数为本表的一行的 types，详见 types.h
+// 创建表，可变参数 list 为本表的一行的 types，详见 types.h
+// list 以 type_t 为单元，遇到 uint64_t ptr 时偏移 +8
 // 如果 types 为外键，需要紧跟一个 uint64_t ptr
 // 指示外键链接到的表位置
 // len(buf) >= 4096+8+2=4106
@@ -43,18 +44,18 @@ uint64_t add_table_index(int fd, void* table, uint16_t pos);
 int remove_table_index(int fd, void* table, uint16_t pos);
 
 // 插入一行，如果 pk 有值则替换
+// list 以 key_t 为单元
 // 如果当前项有 nullable 属性，需要在此项之前
-// 加一个 int isavailable，标记本项是否有值
+// 加一个 key_t isavailable，标记本项是否有值
 // 如果 isavailable==0，后面不再跟有本项数据
 // 如果 isavailable!=0，则在后面附加数据
-// 如果 val 不为 string，直接装填其值
-// 否则，值是指向 string 的指针 (const char*)
-// 如果是 binary，需要在指针之前提供一个 uint32 参数
-// 说明 binary 的大小
+// 如果 val 不为 string/binary，直接装填其值
+// 否则，值是指向 string/binary 的指针 (const char*)
+// 且需要在指针之前提供一个 key_t 参数指示其大小
 // 返回：
 //    0   失败，详见 errno
 //    ptr 本行插入的位置
-uint64_t insert_row(int fd, void* table, int row_len, const void* list);
+uint64_t insert_row(int fd, void* table, const key_t* list);
 
 // 根据主键的匹配值查找行
 // 如果主键不为 string，k 直接装填其值

src/table.c

@@ -17,20 +17,51 @@ static int _calc_index_size(type_t t) {
     switch(t&7) {
         case TYPE_INT8:
             return INT8_INDEX_SZ+10;
+            break;
         case TYPE_INT16:
             return INT16_INDEX_SZ+10 + INT16_BITMAP_SZ+8*2;
+            break;
         case TYPE_INT32:
         case TYPE_FLOAT:
         case TYPE_INT64:
         case TYPE_DOUBLE:
         case TYPE_STRING:
             return PAGESZ+8;
+            break;
         case TYPE_BINARY: // 不能创建索引
         default:
             return EOF;
+            break;
     }
 }
 
+// 计算本类型占用的空间
+static int _calc_type_size(type_t t) {
+    switch(t&7) {
+        case TYPE_INT8:
+            return (int)(!(t&EXTYPE_NONNULL))+1+((t&EXTYPE_UNIQUE)?0:8);
+            break;
+        case TYPE_INT16:
+            return (int)(!(t&EXTYPE_NONNULL))+2+((t&EXTYPE_UNIQUE)?0:8);
+            break;
+        case TYPE_INT32:
+        case TYPE_FLOAT:
+            return (int)(!(t&EXTYPE_NONNULL))+4;
+            break;
+        case TYPE_INT64:
+        case TYPE_DOUBLE:
+            return (int)(!(t&EXTYPE_NONNULL))+8;
+            break;
+        case TYPE_STRING:
+            return (int)(!(t&EXTYPE_NONNULL))+8+2;
+            break;
+        case TYPE_BINARY:
+            return (int)(!(t&EXTYPE_NONNULL))+2;
+            break;
+    }
+    return 0;
+}
+
 // 为 t 类型创建索引，写入 index_ptr
 // 返回：
 //    1  失败，详见 errno
@@ -75,7 +106,8 @@ static int _remove_index_type(int fd, type_t t, uint64_t ptr) {
     return sz;
 }
 
-// 创建表，可变参数为本表的一行的 types，详见 types.h
+// 创建表，可变参数 list 为本表的一行的 types，详见 types.h
+// list 以 type_t 为单元，遇到 uint64_t ptr 时偏移 +8
 // 如果 types 为外键，需要紧跟一个 uint64_t ptr
 // 指示外键链接到的表位置
 // len(buf) >= 4096+8+2=4106
@@ -123,7 +155,7 @@ void* create_table(int fd, char* buf, const char* name, int row_len, const void*
         printf("fill[%d]: %d\n", ap, t);
     #endif
     if(t & EXTYPE_FOREIGNKEY) { // 是外键，还有一个参数
-        ptr = ((uint64_t*)list)[ap];
+        ptr = le64(list+ap);
         #ifdef DEBUG
             printf("fill[%d]: %016llx\n", ap, ptr);
         #endif
@@ -140,7 +172,7 @@ void* create_table(int fd, char* buf, const char* name, int row_len, const void*
             printf("fill[%d]: %d\n", ap, t);
         #endif
         if(t & EXTYPE_FOREIGNKEY) { // 是外键，还有一个参数
-            ptr = ((uint64_t*)list)[ap];
+            ptr = le64(list+ap);
             #ifdef DEBUG
                 printf("fill[%d]: %016llx\n", ap, ptr);
             #endif
@@ -276,16 +308,191 @@ int remove_table_index(int fd, void* table, uint16_t pos) {
     return sync_block(fd, table);
 }
 
+#define DEBUG
+
 // 插入一行，如果 pk 有值则替换
+// list 以 key_t 为单元
 // 如果当前项有 nullable 属性，需要在此项之前
-// 加一个 int isavailable，标记本项是否有值
+// 加一个 key_t isavailable，标记本项是否有值
 // 如果 isavailable==0，后面不再跟有本项数据
 // 如果 isavailable!=0，则在后面附加数据
+// 如果 val 不为 string/binary，直接装填其值
+// 否则，值是指向 string/binary 的指针 (const char*)
+// 且需要在指针之前提供一个 key_t 参数指示其大小
 // 返回：
 //    0   失败，详见 errno
 //    ptr 本行插入的位置
-uint64_t insert_row(int fd, void* table, int row_len, const void* list) {
-    return 0;
+uint64_t insert_row(int fd, void* table, const key_t* list) {
+    int len = 8+2+le16(table+8);
+    int rlen = le16(table+len);
+    int sz = 0; // 本行长度
+    int ap = 0;
+    len += 2;
+
+    uint64_t ptr = *(uint64_t*)(table+len+rlen);
+    int indexsz = _calc_index_size(((type_t*)table)[len]);
+    if(indexsz <= 0) return 0;
+    void* indexbuf = malloc(indexsz);
+    if(!indexbuf) return 0;
+    void* index = load_index(fd, ((type_t*)table)[len], ptr, indexbuf);
+    if(index == NULL) {
+        free(indexbuf);
+        return 0;
+    }
+    ptr = find_item_by_key(fd, ((type_t*)table)[len], index, list[0]);
+
+    for(int i = 0; i < rlen; i++) {
+        type_t t = ((type_t*)table)[len+i];
+        #ifdef DEBUG
+            printf("type: %d, ", (int)t);
+        #endif
+        int isnull = 0;
+        sz += _calc_type_size(t);
+        #ifdef DEBUG
+            printf("sz: %d, ", sz);
+        #endif
+        if(!(t&EXTYPE_NONNULL)) { // 可空，读取是否为空
+            isnull = !((int)(list[ap++]));
+            #ifdef DEBUG
+                printf("isnull: %s, ", isnull?"true":"false");
+            #endif
+        }
+        if((t&7) >= TYPE_STRING) { // 是 string/binary，多读取一个长度
+            int blen = (int)(list[ap++]);
+            #ifdef DEBUG
+                printf("blen: %d, ", blen);
+            #endif
+            if(blen > PAGESZ/2 || blen <= 0) { // 长度超标
+                errno = EFBIG;
+                free(indexbuf);
+                return 0;
+            }
+            sz += blen;
+        }
+        if(!isnull) {
+            #ifdef DEBUG
+                printf("skip key: %lld, ", list[ap]);
+            #endif
+            ap++; // 跳过读取真实值
+        }
+        #ifdef DEBUG
+            printf("ap: %d\n", ap);
+        #endif
+    }
+    #ifdef DEBUG
+        printf("total size: %d\n", sz);
+    #endif
+    if(sz <= 0 || sz > PAGESZ) { // 总长超出一页，无法插入
+        errno = EFBIG;
+        free(indexbuf);
+        return 0;
+    }
+
+    void* buf = malloc(sz+10);
+    if(!buf) {
+        free(indexbuf);
+        return 0;
+    }
+    void* blk = ptr?get_block(fd, sz, ptr, buf):alloc_block(fd, sz, buf);
+    if(blk == NULL) {
+        free(indexbuf);
+        free(buf);
+        return 0;
+    }
+
+    ap = 0;
+    int p = 0;
+    for(int i = 0; i < rlen; i++) {
+        type_t t = ((type_t*)table)[len+i];
+        int blen;
+        if(!(t&EXTYPE_NONNULL)) { // 可空，读取是否为空
+            int isnull = !((int)(list[ap++]));
+            ((uint8_t*)blk)[p++] = isnull;
+            if(isnull) {
+                p += _calc_type_size(t)-1; // 跳过本项
+                if((t&7) >= TYPE_STRING) { // 是 string/binary，多读取一个长度
+                    blen = (int)(list[ap++]);
+                    if(blen > PAGESZ/2 || blen <= 0) { // 长度超标
+                        errno = EFBIG;
+                        free(indexbuf);
+                        return 0;
+                    }
+                    p += blen;
+                }
+                #ifdef DEBUG
+                    printf("skip to: %d\n", p);
+                #endif
+                continue;
+            }
+        }
+        switch(t&7) {
+            case TYPE_INT8:
+                if(!(t&EXTYPE_UNIQUE)) p += 8; // 跳过下一个哈希相同的数据项的指针
+                ((uint8_t*)blk)[p++] = (uint8_t)(list[ap++]);
+                break;
+            case TYPE_INT16:
+                if(!(t&EXTYPE_UNIQUE)) p += 8; // 跳过下一个哈希相同的数据项的指针
+                putle16(blk+p, list[ap++]);
+                p += 2;
+                break;
+            case TYPE_INT32:
+            case TYPE_FLOAT:
+                putle32(blk+p, list[ap++]);
+                p += 4;
+                break;
+            case TYPE_INT64:
+            case TYPE_DOUBLE:
+                putle64(blk+p, list[ap++]);
+                p += 8;
+                break;
+            case TYPE_STRING: // 跳过哈希相同指针
+                p += 8;
+            case TYPE_BINARY: // 是 string/binary，多读取一个长度
+                blen = (int)(list[ap++]);
+                #ifdef DEBUG
+                    printf("blen: %d, ", blen);
+                #endif
+                if(blen > PAGESZ/2 || blen <= 0) { // 长度超标
+                    errno = EFBIG;
+                    free(indexbuf);
+                    return 0;
+                }
+                putle16(blk+p, blen); // 写入长度
+                p += 2;
+                char* field = (char*)(list[ap++]); // 读取指针
+                #ifdef DEBUG
+                    printf("p: %d, ", p);
+                #endif
+                memcpy(blk+p, field, blen); // 复制
+                #ifdef DEBUG
+                    printf("copy field: ");
+                    for(int i = 0; i < blen; i++) printf("%02x ", (int)((unsigned char)(field[i])));
+                    putchar('\n');
+                #endif
+                p += blen;
+                break;
+        }
+    }
+    #ifdef DEBUG
+        printf("total size: %d\n", p);
+    #endif
+
+    if(sync_block(fd, blk)) { // 将行写入文件
+        free(indexbuf);
+        free(buf);
+        return 0;
+    }
+
+    ptr = le64(buf);
+    if(insert_item(fd, ((type_t*)table)[len], index, list[0], ptr)) { // 插入 pk 索引
+        free(indexbuf);
+        free(buf);
+        return 0;
+    }
+    // TODO: 同时插入其它已创建索引的列的索引
+    free(indexbuf);
+    free(buf);
+    return ptr;
 }
 
 // 根据主键的匹配值查找行
@@ -295,7 +502,32 @@ uint64_t insert_row(int fd, void* table, int row_len, const void* list) {
 //    0  失败，详见 errno
 //    ptr  行所在位置
 uint64_t find_row_by_pk(int fd, void* table, key_t k) {
-    return 0;
+    int len = 8+2+le16(table+8);
+    int rlen = le16(table+len);
+    len += 2;
+
+    uint64_t ptr = *(uint64_t*)(table+len+rlen);
+    #ifdef DEBUG
+        printf("indexptr: %016llx, ", ptr);
+    #endif
+    int indexsz = _calc_index_size(((type_t*)table)[len]);
+    #ifdef DEBUG
+        printf("indexsz: %d, ", indexsz);
+    #endif
+    if(indexsz <= 0) return 0;
+    void* indexbuf = malloc(indexsz);
+    if(!indexbuf) return 0;
+    void* index = load_index(fd, ((type_t*)table)[len], ptr, indexbuf);
+    if(index == NULL) {
+        free(indexbuf);
+        return 0;
+    }
+    ptr = find_item_by_key(fd, ((type_t*)table)[len], index, k);
+    #ifdef DEBUG
+        printf("ptr: %016llx\n", ptr);
+    #endif
+    free(indexbuf);
+    return ptr;
 }
 
 // 根据任意匹配值遍历查找行

tests/table_test.c

@@ -18,10 +18,11 @@ int main() {
     }
     if(init_file_header_page(fd)) return 2;
     void* table = create_table(
-        fd, buf, "test_table", 5, (type_t[]){
+        fd, buf, "test_table", 6, (type_t[]){
             TYPE_INT16|EXTYPE_NONNULL|EXTYPE_UNIQUE,
             TYPE_INT64,
             TYPE_INT8,
+            TYPE_INT32,
             TYPE_STRING,
             TYPE_BINARY
         }
@@ -53,5 +54,71 @@ int main() {
     }
     if(strcmp(get_table_name(table, namebuf), "test_table")) return 5;
     if(get_index_ptr(table, 0) != 0x100) return 6;
+    if(insert_row(fd, table, (key_t[]){
+        2333,
+        EXTYPE_NONNULL, 234532121345,
+        EXTYPE_NONNULL, 88,
+        EXTYPE_NULL,
+        EXTYPE_NONNULL, 12, (key_t)"test_string",
+        EXTYPE_NONNULL, 5, (key_t)"\x45\xff\x12\xda"
+    }) == 0) {
+        perror("insert_row");
+        return 7;
+    }
+    uint64_t ptr = find_row_by_pk(fd, table, 2333);
+    if(ptr == 0) {
+        perror("find_row_by_pk");
+        return 8;
+    }
+
+    if(lseek(fd, ptr, SEEK_SET) < 0) return 9;
+    readle16(fd, ptr);
+    if((uint16_t)ptr != 2333) {
+        printf("pk: %d\n", (uint16_t)ptr);
+        return 10;
+    }
+
+    char isnull = 1;
+    read(fd, &isnull, 1);
+    if(isnull) return 11;
+    readle64(fd, ptr);
+    if(ptr != 234532121345) return 12;
+
+    isnull = 1;
+    read(fd, &isnull, 1);
+    if(isnull) return 11;
+    lseek(fd, 8, SEEK_CUR);
+    read(fd, &isnull, 1);
+    if(isnull != 88) return 13;
+
+    isnull = 0;
+    read(fd, &isnull, 1);
+    if(!isnull) return 11;
+    lseek(fd, 4, SEEK_CUR);
+
+    isnull = 1;
+    read(fd, &isnull, 1);
+    if(isnull) return 11;
+    lseek(fd, 8, SEEK_CUR);
+    readle16(fd, ptr);
+    if((uint16_t)ptr != 12) {
+        printf("string len: %d\n", (uint16_t)ptr);
+        return 10;
+    }
+    char buf[20];
+    read(fd, buf, 12);
+    if(strcmp(buf, "test_string")) return 14;
+
+    isnull = 1;
+    read(fd, &isnull, 1);
+    if(isnull) return 11;
+    readle16(fd, ptr);
+    if((uint16_t)ptr != 5) {
+        printf("binary len: %d\n", (uint16_t)ptr);
+        return 10;
+    }
+    read(fd, buf, 5);
+    if(strcmp(buf, "\x45\xff\x12\xda")) return 15;
     close(fd);
+    // remove("table_test_tmp.bin");
 }