8.行与列

# 08. 行与列

# 8.1 引言

在键值对存储（KV store）之上构建关系型数据库的第一步是添加表。表由一系列的行和列组成。一部分列被定义为 “主键”；主键具有唯一性，因此可以在查询和二级索引中用来引用某一行数据。

表如何融入键值对存储呢？我们可以将一行数据拆分为两部分：

  1. 主键列存入键值对存储的 “键” 部分。
  2. 非主键列存入键值对存储的 “值” 部分。

这样我们就可以对主键进行点查询和范围查询。目前，我们只考虑对主键的查询，二级索引的使用将在后续章节介绍。

# 8.2 数据结构

以下是行和单元格的定义。目前，我们只支持两种数据类型（int64和字节切片）。

const  (
    TYPE_ERROR =  0
    TYPE_BYTES =  1
    TYPE_INT64 =  2
)

// 表单元格
type  Value struct  {
    Type uint32
    I64  int64
    Str  []byte
}

// 表行
type  Record struct  {
    Cols []string
    Vals []Value
}

func  ( rec *Record)  AddStr(key string,  val []byte)  *Record
func  ( rec *Record)  AddInt64(key string,  val int64)  *Record
func  ( rec *Record)  Get(key string)  *Value
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

数据库（DB）和表的定义如下：

type  DB struct  {
    Path string
    // 内部使用
    kv     KV
    tables map[string]*TableDef // 缓存的表定义
}

// 表定义
type  TableDef struct  {
    // 用户定义
    Name  string
    Types []uint32 // 列类型
    Cols  []string // 列名
    PKeys int     // 前PKeys列是主键
    // 为不同表自动分配的B树键前缀
    Prefix uint32
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

为了支持多个表，键值对存储中的键会加上一个唯一的32位数字作为前缀。

表定义需要存储在某个地方，我们将使用一个内部表来存储它们。同时，我们还会添加一个内部表来存储数据库自身使用的元数据。

// 内部表：元数据
var  TDEF_META =  &TableDef{
    Prefix:  1,
    Name:       "@meta",
    Types:     []uint32{TYPE_BYTES,  TYPE_BYTES},
    Cols:       []string{"key",  "val"},
    PKeys:     1,
}

// 内部表：表结构
var  TDEF_TABLE =  &TableDef{
    Prefix:  2,
    Name:       "@table",
    Types:     []uint32{TYPE_BYTES,  TYPE_BYTES},
    Cols:       []string{"name",  "def"},
    PKeys:     1,
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 8.3 点查询

我们来实现基于主键的点查询，范围查询将在下一章添加。

// 根据主键获取单行数据
func  dbGet(db *DB,  tdef *TableDef,  rec *Record)   (bool,  error)  {
    values,  err :=  checkRecord(tdef,  * rec,  tdef.PKeys)
    if  err !=  nil {
        return  false,  err
    }

    key :=  encodeKey(nil,  tdef.Prefix,  values[:tdef.PKeys])
    val,  ok :=  db.kv.Get(key)
    if   ! ok {
        return  false,  nil
    }

    for  i :=  tdef.PKeys;  i <  len(tdef.Cols);  i++  {
        values[i].Type =  tdef.Types[i]
    }
    decodeValues(val,  values[tdef.PKeys:])

    rec.Cols =  append(rec.Cols,  tdef.Cols[tdef.PKeys:]... )
    rec.Vals =  append(rec.Vals,  values[tdef.PKeys:]... )
    return  true,  nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

具体步骤如下：

  1. 验证输入是否为完整的主键。（checkRecord）
  2. 对主键进行编码。（encodeKey）
  3. 在键值对存储中查询。（db.kv.Get）
  4. 对值进行解码。（decodeValues）
// 重新排序记录并检查是否缺少列。
// n == tdef.PKeys：记录恰好是主键
// n == len(tdef.Cols)：记录包含所有列
func  checkRecord(tdef *TableDef,  rec Record,  n int)   ([]Value,  error)  {
    // 省略……
}
1
2
3
4
5
6

数据编码为字节和从字节解码的方法将在下一章解释。目前，本章中任何序列化方案都可以。

func  encodeValues(out []byte,  vals []Value)   []byte
func  decodeValues(in []byte,  out []Value)

// 用于主键
func  encodeKey(out []byte,  prefix uint32,  vals []Value)   []byte {
    var  buf [4]byte
    binary.BigEndian.PutUint32(buf[:],  prefix)
    out =  append(out,  buf[:]... )
    out =  encodeValues(out,  vals)
    return  out
}
1
2
3
4
5
6
7
8
9
10
11

要查询一个表，我们必须首先获取其定义。

// 根据主键获取单行数据
func  (db *DB)  Get(table string,  rec *Record)  (bool,  error)  {
    tdef :=  getTableDef(db,  table)
    if  tdef ==  nil {
        return  false,  fmt.Errorf("table not found: %s",  table)
    }
    return  dbGet(db,  tdef,  rec)
}
1
2
3
4
5
6
7
8

表定义以JSON格式存储在内部表TDEF_TABLE中。

// 根据表名获取表定义
func  getTableDef(db *DB,  name string)  *TableDef {
    tdef,  ok :=  db.tables[name]
    if   ! ok {
        if  db.tables ==  nil {
            db.tables =  map[string]*TableDef{}
        }
        tdef =  getTableDefDB(db,  name)
        if  tdef !=  nil {
            db.tables[name]  =  tdef
        }
    }
    return  tdef
}

func  getTableDefDB(db *DB,  name string)  *TableDef {
    rec :=   (&Record{}).AddStr("name",   []byte(name))
    ok,  err :=  dbGet(db,  TDEF_TABLE,   rec)
    assert(err ==  nil)
    if   ! ok {
        return  nil
    }

    tdef :=  &TableDef{}
    err =  json.Unmarshal(rec.Get("def").Str,  tdef)
    assert(err ==  nil)
    return  tdef
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 8.4 更新操作

更新操作可以是插入新行或替换现有行。B树接口被修改以支持不同的更新模式。

// 更新模式
const  (
    MODE_UPSERT     =  0 // 插入或替换
    MODE_UPDATE_ONLY =  1 // 仅更新现有键
    MODE_INSERT_ONLY =  2 // 仅添加新键
)

type  InsertReq struct  {
    tree *BTree
    // 输出
    Added bool // 是否添加了新键
    // 输入
    Key []byte
    Val []byte
    Mode int
}

func  (tree *BTree)  InsertEx(req *InsertReq)
func  (db *KV)  Update(key []byte,  val []byte,  mode int)   (bool,  error)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

通过主键更新记录的函数：

// 向表中添加一行数据
func  dbUpdate(db *DB,  tdef *TableDef,   rec Record,  mode int)   (bool,  error)  {
    values,  err :=  checkRecord(tdef,   rec,  len(tdef.Cols))
    if  err !=  nil {
        return  false,  err
    }

    key :=  encodeKey(nil,  tdef.Prefix,  values[:tdef.PKeys])
    val :=  encodeValues(nil,  values[tdef.PKeys:])
    return  db.kv.Update(key,  val,  mode)
}
1
2
3
4
5
6
7
8
9
10
11

不同的更新模式：

// 添加一条记录
func  (db *DB)  Set(table string,   rec Record,  mode int)   (bool,  error)  {
    tdef :=  getTableDef(db,  table)
    if  tdef ==  nil {
        return  false,  fmt.Errorf("table not found: %s",  table)
    }
    return  dbUpdate(db,  tdef,  rec,  mode)
}
func  (db *DB)  Insert(table string,   rec Record)  (bool,  error)  {
    return  db.Set(table,  rec,  MODE_INSERT_ONLY)
}
func  (db *DB)  Update(table string,   rec Record)  (bool,  error)  {
    return  db.Set(table,  rec,  MODE_UPDATE_ONLY)
}
func  (db *DB)  Upsert(table string,   rec Record)  (bool,  error)  {
    return  db.Set(table,  rec,  MODE_UPSERT)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

删除一行数据的操作类似：

// 根据主键删除一条记录
func  dbDelete(db *DB,  tdef *TableDef,   rec Record)  (bool,  error)  {
    values,  err :=  checkRecord(tdef,   rec,  tdef.PKeys)
    if  err !=  nil {
        return  false,  err
    }

    key :=  encodeKey(nil,  tdef.Prefix,  values[:tdef.PKeys])
    return  db.kv.Del(key)
}

func  (db *DB)  Delete(table string,   rec Record)  (bool,  error)  {
    tdef :=  getTableDef(db,  table)
    if  tdef ==  nil {
        return  false,  fmt.Errorf("table not found: %s",  table)
    }
    return  dbDelete(db,  tdef,  rec)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 8.5 创建新表

创建新表需要三个步骤：

  1. 检查表定义。
  2. 分配表键前缀。
  3. 存储下一个表前缀和表定义。
func  (db *DB)  TableNew(tdef *TableDef)  error {
    if  err :=  tableDefCheck(tdef);  err !=  nil {
        return  err
    }

    // 检查已存在的表
    table :=   (&Record{}).AddStr("name",   []byte(tdef.Name))
    ok,  err :=  dbGet(db,  TDEF_TABLE,  table)
    assert(err ==  nil)
    if  ok {
        return  fmt.Errorf("table exists: %s",  tdef.Name)
    }

    // 分配新的前缀
    assert(tdef.Prefix ==  0)
    tdef.Prefix =  TABLE_PREFIX_MIN
    meta :=   (&Record{}).AddStr("key",   []byte("next_prefix"))
    ok,  err =  dbGet(db,  TDEF_META,  meta)
    assert(err ==  nil)
    if  ok {
        tdef.Prefix =  binary.LittleEndian.Uint32(meta.Get("val").Str)
        assert(tdef.Prefix >  TABLE_PREFIX_MIN)
    }  else  {
        meta.AddStr("val",  make([]byte,  4))
    }

    // 更新下一个前缀
    binary.LittleEndian.PutUint32(meta.Get("val").Str,  tdef.Prefix+1)
    _,  err =  dbUpdate(db,  TDEF_META,  *meta,  0)
    if  err !=  nil {
        return  err
    }

    // 存储表定义
    val,  err :=  json.Marshal(tdef)
    assert(err ==  nil)
    table.AddStr("def",  val)
    _,  err =  dbUpdate(db,  TDEF_TABLE,  *table,  0)
    return  err
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

前缀数字从TDEF_META内部表的next_prefix键开始依次分配。表定义以JSON格式存储在TDEF_TABLE表中。

尽管我们添加了表结构，但结果在很大程度上仍然是一个键值对存储。还缺少一些重要的方面：

  1. 下一章的范围查询。
  2. 再下一章的二级索引。
上次更新: 2025/04/16, 02:04:00
7.空闲列表：重用页面
9.范围查询

社区交流

讨论与留言

前往 GitHub Issues →

本留言区仅对应当前文章，欢迎补充观点、提出问题或帮助修正文中疏漏。 留言由 GitHub/Gitalk 提供，需要使用 GitHub 登录。

最近更新
第 1 课：专栏导论与安全边界
07-01
第 3 课：安装编译与调试环境
07-01
第 4 课：第三方库下载、编译与依赖管理
07-01
更多文章>

Copyright © 2024-2026 沪ICP备2023015129号 张小方 版权所有