功能:增强 Schema 系统和添加新示例
- 扩展 Schema 支持更多数据类型(Duration、URL、JSON 等) - 优化 SSTable 编码解码性能 - 添加多个新示例程序: - all_types: 展示所有支持的数据类型 - new_types: 演示新增类型的使用 - struct_tags: 展示结构体标签功能 - time_duration: 时间和持续时间处理示例 - 完善测试用例和文档 - 优化代码结构和错误处理
This commit is contained in:
42
table.go
42
table.go
@@ -398,8 +398,8 @@ func (t *Table) insertSingle(data map[string]any) error {
|
||||
Data: data,
|
||||
}
|
||||
|
||||
// 3. 序列化
|
||||
rowData, err := json.Marshal(row)
|
||||
// 3. 序列化(使用二进制格式,保留类型信息)
|
||||
rowData, err := encodeSSTableRowBinary(row, t.schema)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -437,12 +437,12 @@ func (t *Table) Get(seq int64) (*SSTableRow, error) {
|
||||
// 1. 先查 MemTable Manager (Active + Immutables)
|
||||
data, found := t.memtableManager.Get(seq)
|
||||
if found {
|
||||
var row SSTableRow
|
||||
err := json.Unmarshal(data, &row)
|
||||
// 使用二进制解码
|
||||
row, err := decodeSSTableRowBinary(data, t.schema)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &row, nil
|
||||
return row, nil
|
||||
}
|
||||
|
||||
// 2. 查询 SST 文件
|
||||
@@ -454,24 +454,12 @@ func (t *Table) GetPartial(seq int64, fields []string) (*SSTableRow, error) {
|
||||
// 1. 先查 MemTable Manager (Active + Immutables)
|
||||
data, found := t.memtableManager.Get(seq)
|
||||
if found {
|
||||
var row SSTableRow
|
||||
err := json.Unmarshal(data, &row)
|
||||
// 使用二进制解码(支持部分解码)
|
||||
row, err := decodeSSTableRowBinaryPartial(data, t.schema, fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// MemTable 中的数据已经完全解析,需要手动过滤字段
|
||||
if len(fields) > 0 {
|
||||
filteredData := make(map[string]any)
|
||||
for _, field := range fields {
|
||||
if val, ok := row.Data[field]; ok {
|
||||
filteredData[field] = val
|
||||
}
|
||||
}
|
||||
row.Data = filteredData
|
||||
}
|
||||
|
||||
return &row, nil
|
||||
return row, nil
|
||||
}
|
||||
|
||||
// 2. 查询 SST 文件(按需解码)
|
||||
@@ -505,10 +493,10 @@ func (t *Table) flushImmutable(imm *ImmutableMemTable, walNumber int64) error {
|
||||
var rows []*SSTableRow
|
||||
iter := imm.NewIterator()
|
||||
for iter.Next() {
|
||||
var row SSTableRow
|
||||
err := json.Unmarshal(iter.Value(), &row)
|
||||
// 使用二进制解码
|
||||
row, err := decodeSSTableRowBinary(iter.Value(), t.schema)
|
||||
if err == nil {
|
||||
rows = append(rows, &row)
|
||||
rows = append(rows, row)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -609,10 +597,10 @@ func (t *Table) recover() error {
|
||||
|
||||
// 重放 WAL 到 Active MemTable
|
||||
for _, entry := range entries {
|
||||
// 验证 Schema
|
||||
var row SSTableRow
|
||||
if err := json.Unmarshal(entry.Data, &row); err != nil {
|
||||
return fmt.Errorf("failed to unmarshal row during recovery (seq=%d): %w", entry.Seq, err)
|
||||
// 使用二进制解码验证 Schema
|
||||
row, err := decodeSSTableRowBinary(entry.Data, t.schema)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to decode row during recovery (seq=%d): %w", entry.Seq, err)
|
||||
}
|
||||
|
||||
// 验证 Schema
|
||||
|
||||
Reference in New Issue
Block a user