fix: 复合索引真正生效 (键按分量定长拼接 + 复合区间推导)

- extractIndexKey: 按 FieldIDs 顺序拼接各分量 (原先只取 FieldIDs[0]), 写入/删除/更新/批量重建共用同一编码

- query: 把 WHERE 折叠成[等值前缀 + 一个范围分量]的键区间, 残余条件下置过滤; 单列索引是 n=1 的特例

- CreateIndex: 拒绝重复列 (会让索引永远取不齐分量而静默为空); 行存储索引(主键/时序)拒绝复合键

- 新增 Table.BuildIndexKey 公开构造复合键; encodeIndexKeySafe 把类型不符的 panic 转成 error

- ORM: 同名字段声明的 index tag 合并为复合索引; 已有索引分量数不符时直接报错而非静默沿用

- 测试: core 端到端 (页级 EntryLen 证据 + 等值/最左前缀/范围/严格边界/更新/删除/重启/批量重建) 与 ORM 合并校验

DB/core/IndexMetaPage.go

@@ -118,6 +118,24 @@ func (im *IndexMetaPage) CreateIndex(indexName string, itype page.IndexType, fie
 		return nil, ErrIndexFieldCount
 	}
 
+	// 复合键要求各分量互不相同。重复列会让 extractIndexKey 永远取不齐分量
+	// (某一分量无人填充), 索引静默变成空索引 —— 写入成功却查不到, 因此直接拒绝。
+	var seen [256]bool
+	for _, fid := range fieldIDs {
+		if seen[fid] {
+			return nil, ErrIndexFieldDuplicated
+		}
+		seen[fid] = true
+	}
+
+	// 行存储索引 (主键 B+ 树 / 时序) 的键同时就是"行在哪里"的定位键,
+	// 复合键会让回表语义失效 (ResolveByPK 用 pkID, 与键是两回事), 明确拒绝。
+	if fc > 1 {
+		if h, ok := index.Compose.GetHandler(itype); ok && h.IsRowStore() {
+			return nil, ErrIndexCompositeRowStore
+		}
+	}
+
 	for _, m := range im.metas {
 		if util.BytesToString(m.IndexName[:]) == indexName && m.Flag&0x80 == 0 {
 			return nil, ErrIndexExists
@@ -313,11 +331,22 @@ func (im *IndexMetaPage) saveMeta(m *IndexMeta) {
 
 // extractIndexKey 从行数据中提取索引键 (版本感知)
 // rowData 格式: [schemaVer(1B)] [field_data...]
+//
+// 复合索引的键 = 按 FieldIDs 顺序, 把各分量字段的原始字节定长拼接。
+//   - 顺序取 FieldIDs 而非行内物理顺序 —— 查询侧据此才能做"列前缀"范围推导:
+//     (a, b) 上的索引既能服务 a=?, 也能服务 a=? AND b>?;
+//   - 每个分量长度 = GetSize(字段类型), 与行内编码完全一致, 因此拼接键是定长的,
+//     字节序比较等价于"先比 a 再比 b"的字典序;
+//   - 任一分量在该行版本里取不到 (列被 Drop / schema 版本不匹配) → 返回 nil,
+//     该索引跳过这一行, 与单列索引"取不到键就跳过"的既有语义一致。
 func extractIndexKey(rowData []byte, meta *IndexMeta, fields []types.FieldInfo, activeVer byte, versions []SchemaVersion) []byte {
-	if meta.FieldCnt == 0 || len(rowData) < 1 {
+	if meta == nil || meta.FieldCnt == 0 || len(rowData) < 1 {
+		return nil
+	}
+	n := int(meta.FieldCnt)
+	if n > len(meta.FieldIDs) {
 		return nil
 	}
-	fieldIdx := int(meta.FieldIDs[0])
 
 	// 从 rowData 读取该行的 schema 版本
 	ver := rowData[0]
@@ -332,21 +361,49 @@ func extractIndexKey(rowData []byte, meta *IndexMeta, fields []types.FieldInfo,
 		return nil
 	}
 
-	// 按该版本的字段顺序计算偏移
-	off := 1 // 跳过 schemaVer
+	// 沿该版本的字段顺序走一遍行数据, 顺手取齐各分量。
+	// seg[k] 指向 rowData 的子切片, 不做拷贝; 全部取齐后再一次性拼出键。
+	var (
+		seg  [16][]byte
+		off  = 1 // 跳过 schemaVer
+		need = n
+	)
 	for _, idx := range sv.FieldIndices {
-		if int(idx) == fieldIdx {
-			keyLen := int(getFieldSize(fields[fieldIdx].FieldType))
-			if off+keyLen > len(rowData) {
+		fi := int(idx)
+		if fi >= len(fields) {
+			return nil
+		}
+		size := int(getFieldSize(fields[fi].FieldType))
+		for k := 0; k < n; k++ {
+			if int(meta.FieldIDs[k]) != fi {
+				continue
+			}
+			if off+size > len(rowData) {
 				return nil
 			}
-			key := make([]byte, keyLen)
-			copy(key, rowData[off:off+keyLen])
-			return key
+			seg[k] = rowData[off : off+size]
+			need--
+			break
+		}
+		off += size
+		if need == 0 {
+			break // 分量已取齐, 后面的字段与本索引无关
 		}
-		off += int(getFieldSize(fields[idx].FieldType))
 	}
-	return nil
+	if need != 0 {
+		return nil // 有分量不在该版本中 (如该列已被 Drop), 无法构成完整键
+	}
+
+	total := 0
+	for k := 0; k < n; k++ {
+		total += len(seg[k])
+	}
+	key := make([]byte, 0, total)
+	for k := 0; k < n; k++ {
+		key = append(key, seg[k]...)
+	}
+	_ = activeVer
+	return key
 }
 
 // InsertRow 插入行到所有索引

DB/core/Table.go

@@ -6,6 +6,7 @@ import (
 	"TsDB2/DB/core/util"
 	"bytes"
 	"encoding/binary"
+	"fmt"
 	"strings"
 	"unsafe"
 )
@@ -187,6 +188,46 @@ func (table *Table) CreateIndex(idxName string, typ page.IndexType, fields []byt
 }
 func (table *Table) DropIndex(idx string) error { return table.indexMeta.DropIndex(idx) }
 
+// BuildIndexKey 按索引定义, 把各分量值编码成索引键 (与写入侧 extractIndexKey 同一编码)。
+//
+// 复合索引的键是各分量定长拼接, 外部无法凭空构造 —— 因此点查/删除/更新复合索引
+// (SearchByIndex / DeleteRow / UpdateRow) 都要先经这里把值编成键。
+// values 的顺序与建索引时给的 fieldIDs 顺序一致, 个数必须相等。
+func (table *Table) BuildIndexKey(idxName string, values ...interface{}) ([]byte, error) {
+	m := table.indexMeta.findIndexMetaByName(idxName)
+	if m == nil {
+		return nil, ErrIndexNotFound
+	}
+	n := int(m.FieldCnt)
+	if n != len(values) {
+		return nil, fmt.Errorf("索引 %q 需要 %d 个分量值, 实际给了 %d 个", idxName, n, len(values))
+	}
+	key := make([]byte, 0, n*8)
+	for k := 0; k < n; k++ {
+		fi := int(m.FieldIDs[k])
+		if fi >= len(table.Field) {
+			return nil, ErrFieldNotFound
+		}
+		ft := table.Field[fi].FieldType
+		if ft&0x80 != 0 {
+			return nil, fmt.Errorf("索引 %q 的第 %d 个分量列已被删除", idxName, k+1)
+		}
+		if types.IsDynamic(ft) {
+			return nil, fmt.Errorf("索引 %q 的第 %d 个分量是动态列, 不能作为索引键", idxName, k+1)
+		}
+		enc, err := encodeIndexKeySafe(values[k], ft)
+		if err != nil {
+			return nil, fmt.Errorf("索引 %q 的第 %d 个分量: %w", idxName, k+1, err)
+		}
+		if want := int(getFieldSize(ft)); len(enc) != want {
+			return nil, fmt.Errorf("索引 %q 的第 %d 个分量: 编码 %d 字节, 列宽 %d 字节",
+				idxName, k+1, len(enc), want)
+		}
+		key = append(key, enc...)
+	}
+	return key, nil
+}
+
 // TotalRows 返回表的总行数
 func (table *Table) TotalRows() uint64 { return table.info.TotalRows }
 

DB/core/errors.go

@@ -1,4 +1,4 @@
-package core
+package core
 
 import "errors"
 
@@ -9,15 +9,17 @@ var (
 	ErrTableExists       = errors.New("表已存在")
 	ErrInvalidFieldCount = errors.New("字段数量非法")
 
-	ErrFieldAllocateTooLarge = errors.New("字段分配空间过大")
-	ErrIndexFieldCount       = errors.New("索引字段数量必须在 1-16 之间")
-	ErrIndexExists           = errors.New("索引已存在")
-	ErrFieldNotFound         = errors.New("字段未找到")
-	ErrFieldExists           = errors.New("字段已存在")
-	ErrFieldSpaceNotEnough   = errors.New("字段扩展空间不足，且无法分配新页") // 字段扩展空间不足，且无法分配新页
-	ErrPageRead              = errors.New("当前页不存在")           // 当前页不存在
-	ErrHeapPageTooLarge      = errors.New("申请空间超过单页最大容量")     // 申请空间超过单页最大容量
-	ErrHeapAllocFailed       = errors.New("堆页分配失败")           // 堆页分配失败
-	ErrNotFound              = errors.New("键未找到")             // 键未找到
-	ErrIndexNotFound         = errors.New("索引未找到")            // 索引未找到
+	ErrFieldAllocateTooLarge  = errors.New("字段分配空间过大")
+	ErrIndexFieldCount        = errors.New("索引字段数量必须在 1-16 之间")
+	ErrIndexFieldDuplicated   = errors.New("索引字段重复: 同一列在一个索引里只能出现一次")
+	ErrIndexCompositeRowStore = errors.New("主键/时序这类行存储索引不支持复合键")
+	ErrIndexExists            = errors.New("索引已存在")
+	ErrFieldNotFound          = errors.New("字段未找到")
+	ErrFieldExists            = errors.New("字段已存在")
+	ErrFieldSpaceNotEnough    = errors.New("字段扩展空间不足，且无法分配新页") // 字段扩展空间不足，且无法分配新页
+	ErrPageRead               = errors.New("当前页不存在")           // 当前页不存在
+	ErrHeapPageTooLarge       = errors.New("申请空间超过单页最大容量")     // 申请空间超过单页最大容量
+	ErrHeapAllocFailed        = errors.New("堆页分配失败")           // 堆页分配失败
+	ErrNotFound               = errors.New("键未找到")             // 键未找到
+	ErrIndexNotFound          = errors.New("索引未找到")            // 索引未找到
 )

DB/core/query.go

@@ -6,6 +6,7 @@ import (
 	"TsDB2/DB/core/types"
 	"bytes"
 	"encoding/binary"
+	"fmt"
 	"sort"
 )
 
@@ -82,6 +83,18 @@ func encodeIndexKey(value interface{}, ft types.FieldType) []byte {
 	return buf
 }
 
+// encodeIndexKeySafe 在 encodeIndexKey 外面包一层 panic 恢复。
+// encodeIndexKey 直接对值做类型断言 (value.(uint64) 等), 类型不符会 panic;
+// 内部调用点的值来自条件对象 (受控), 而公开 API 传进来的值不可信, 必须转成 error。
+func encodeIndexKeySafe(value interface{}, ft types.FieldType) (enc []byte, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			enc, err = nil, fmt.Errorf("值 %v 与存储类型 %d 不匹配: %v", value, ft, r)
+		}
+	}()
+	return encodeIndexKey(value, ft), nil
+}
+
 // ======================== 索引匹配 ========================
 
 // fieldNameFromMeta 获取索引第一个字段的名称
@@ -93,85 +106,224 @@ func (q *Query) fieldNameFromMeta(meta *IndexMeta) string {
 	return ""
 }
 
-// fieldTypeFromMeta 获取索引第一个字段的类型
-func (q *Query) fieldTypeFromMeta(meta *IndexMeta) types.FieldType {
-	fieldIdx := int(meta.FieldIDs[0])
-	if fieldIdx < len(q.table.Field) {
-		return q.table.Field[fieldIdx].FieldType
+// ======================== 复合索引键规划 ========================
+//
+// 索引键 = 各分量字段定长拼接 (编码与 extractIndexKey 完全一致), 因此:
+//   - 已知前 k 个分量等值 → 键上一段连续区间;
+//   - 第 k+1 个分量上的范围条件 → 再收窄这段区间;
+//   - 更靠后的分量索引帮不上忙, 原样留作行级后置过滤。
+// 下面的代码把 WHERE 按这个模型折叠成 (startKey, endKey, 残余条件)。
+//
+// 单列索引是 n == 1 的特例, 行为与旧实现一致。
+
+// indexKeyLayout 索引键各分量的偏移与长度 (定长拼接)
+type indexKeyLayout struct {
+	offsets []int
+	sizes   []int
+	total   int
+}
+
+// cmpKind 叶子条件能为键分量提供的边界方向
+type cmpKind int
+
+const (
+	cmpEq cmpKind = iota
+	cmpGe
+	cmpGt
+	cmpLe
+	cmpLt
+)
+
+// keyConstraint 某个键分量上的合并边界 (多个 AND 子条件取交集)
+type keyConstraint struct {
+	lo, hi       []byte // 编码后的分量值; nil 表示该侧无约束
+	loInc, hiInc bool
+	count        int // 落在该分量上的条件个数, 0 = 无约束
+}
+
+// add 把一个边界并入本分量; 返回 false 表示条件自相矛盾。
+func (kc *keyConstraint) add(kind cmpKind, enc []byte) bool {
+	switch kind {
+	case cmpEq:
+		if !kc.addLower(enc, true) || !kc.addUpper(enc, true) {
+			return false
+		}
+	case cmpGe:
+		if !kc.addLower(enc, true) {
+			return false
+		}
+	case cmpGt:
+		if !kc.addLower(enc, false) {
+			return false
+		}
+	case cmpLe:
+		if !kc.addUpper(enc, true) {
+			return false
+		}
+	case cmpLt:
+		if !kc.addUpper(enc, false) {
+			return false
+		}
 	}
-	return 0
+	kc.count++
+	return true
 }
 
-// condToRange 尝试从条件中提取扫描范围, 返回 (startKey, endKey, startInc, endInc, ok)
-func (q *Query) condToRange(cond Condition, fieldName string, ft types.FieldType) ([]byte, []byte, bool, bool, Condition) {
-	switch c := cond.(type) {
-	case EqCondition:
-		if c.Field == fieldName {
-			key := encodeIndexKey(c.Value, ft)
-			return key, key, true, true, nil
+func (kc *keyConstraint) addLower(enc []byte, inclusive bool) bool {
+	if kc.lo == nil {
+		kc.lo, kc.loInc = enc, inclusive
+		return true
+	}
+	switch c := bytes.Compare(enc, kc.lo); {
+	case c > 0: // 取更高的下界
+		kc.lo, kc.loInc = enc, inclusive
+	case c == 0: // 同值: 两端都闭才算"含"
+		kc.loInc = kc.loInc && inclusive
+	}
+	return kc.notContradictory()
+}
+
+func (kc *keyConstraint) addUpper(enc []byte, inclusive bool) bool {
+	if kc.hi == nil {
+		kc.hi, kc.hiInc = enc, inclusive
+		return true
+	}
+	switch c := bytes.Compare(enc, kc.hi); {
+	case c < 0: // 取更低的上界
+		kc.hi, kc.hiInc = enc, inclusive
+	case c == 0:
+		kc.hiInc = kc.hiInc && inclusive
+	}
+	return kc.notContradictory()
+}
+
+// notContradictory 判断 [lo, hi] 是否还可能有值。
+//
+// 注意: 这里按编码字节比较, 而定长整数用的是补码 —— 负数的字节序与数值序相反。
+// 所以它只在"字节序下已确定矛盾"时才返回 false, 此时真值也必然为空, 反向不成立。
+// 调用方因此只把它当作"放弃使用索引"的依据, 绝不据此断言区间非空。
+func (kc *keyConstraint) notContradictory() bool {
+	if kc.lo == nil || kc.hi == nil {
+		return true
+	}
+	switch c := bytes.Compare(kc.lo, kc.hi); {
+	case c > 0:
+		return false
+	case c == 0:
+		return kc.loInc && kc.hiInc
+	}
+	return true
+}
+
+// isEquality 该分量已被钉死为单值 (lo == hi 且两端都闭)
+func (kc *keyConstraint) isEquality() bool {
+	return kc.count > 0 && kc.lo != nil && kc.hi != nil &&
+		kc.loInc && kc.hiInc && bytes.Equal(kc.lo, kc.hi)
+}
+
+// buildKeyLayout 依据索引元信息与表字段类型算出键布局。
+// 返回 false 表示该索引当前不可用于查询:
+//   - 分量列不存在或已被 Drop (新行提不出该分量, 索引里根本没有这批键);
+//   - 分量是动态列 (行内只存 6B 堆引用, 值 != 引用, 无法按键定位)。
+func (q *Query) buildKeyLayout(meta *IndexMeta) (*indexKeyLayout, bool) {
+	n := int(meta.FieldCnt)
+	if n <= 0 || n > len(meta.FieldIDs) || n > len(q.table.Field) {
+		return nil, false
+	}
+	lay := &indexKeyLayout{offsets: make([]int, n), sizes: make([]int, n)}
+	off := 0
+	for k := 0; k < n; k++ {
+		fi := int(meta.FieldIDs[k])
+		if fi >= len(q.table.Field) {
+			return nil, false
+		}
+		ft := q.table.Field[fi].FieldType
+		if ft&0x80 != 0 {
+			return nil, false
+		}
+		if types.IsDynamic(ft) {
+			return nil, false
+		}
+		size := int(getFieldSize(ft))
+		if size <= 0 {
+			return nil, false
+		}
+		lay.offsets[k] = off
+		lay.sizes[k] = size
+		off += size
+	}
+	lay.total = off
+	return lay, true
+}
+
+// locateComponent 返回字段 master 下标在索引键中的分量位置; -1 表示不属于该索引
+func (q *Query) locateComponent(fieldName string, meta *IndexMeta) int {
+	fi := q.table.findFieldIndex(fieldName)
+	if fi < 0 {
+		return -1
+	}
+	n := int(meta.FieldCnt)
+	if n > len(meta.FieldIDs) {
+		n = len(meta.FieldIDs)
+	}
+	for k := 0; k < n; k++ {
+		if int(meta.FieldIDs[k]) == fi {
+			return k
 		}
+	}
+	return -1
+}
+
+// classifyLeaf 判断一个叶子条件能否成为某个键分量的边界。
+// comp < 0 表示与该索引无关 (字段不在索引里, 或该条件类型给不出可比边界)。
+func (q *Query) classifyLeaf(c Condition, meta *IndexMeta) (comp int, value interface{}, kind cmpKind) {
+	switch t := c.(type) {
+	case EqCondition:
+		return q.locateComponent(t.Field, meta), t.Value, cmpEq
 	case GeCondition:
-		if c.Field == fieldName {
-			return encodeIndexKey(c.Value, ft), nil, true, false, nil
-		}
+		return q.locateComponent(t.Field, meta), t.Value, cmpGe
 	case GtCondition:
-		if c.Field == fieldName {
-			return encodeIndexKey(c.Value, ft), nil, false, false, nil
-		}
+		return q.locateComponent(t.Field, meta), t.Value, cmpGt
 	case LeCondition:
-		if c.Field == fieldName {
-			return nil, encodeIndexKey(c.Value, ft), false, true, nil
-		}
+		return q.locateComponent(t.Field, meta), t.Value, cmpLe
 	case LtCondition:
-		if c.Field == fieldName {
-			return nil, encodeIndexKey(c.Value, ft), false, false, nil
-		}
-	case AndCondition:
-		// 合并所有子条件的范围
-		var sk, ek []byte
-		var si, ei bool = true, true // 默认全包含
-		var hasStart, hasEnd bool
-		var remaining []Condition
-
-		for _, sub := range c.Conditions {
-			s, e, sinc, einc, rem2 := q.condToRange(sub, fieldName, ft)
-			if s != nil || e != nil {
-				// 该子条件能为索引提供边界: 合并到已有范围
-				if s != nil {
-					if !hasStart || (sinc && bytes.Compare(s, sk) > 0) || (!sinc && bytes.Compare(s, sk) >= 0) {
-						sk = s
-						si = sinc
-						hasStart = true
-					}
-				}
-				if e != nil {
-					if !hasEnd || (einc && bytes.Compare(e, ek) < 0) || (!einc && bytes.Compare(e, ek) <= 0) {
-						ek = e
-						ei = einc
-						hasEnd = true
-					}
-				}
-				// 未被完全吸收的残余条件仍需在扫描结果上后置过滤
-				if rem2 != nil {
-					remaining = append(remaining, rem2)
-				}
-			} else {
-				// 此子条件不能用索引: 整体留作后置过滤
-				remaining = append(remaining, sub)
-			}
+		return q.locateComponent(t.Field, meta), t.Value, cmpLt
+	default:
+		return -1, nil, cmpEq
+	}
+}
+
+// flattenAnd 把 AND 树展平成叶子条件。
+// OR / NOT / LIKE / IN 不是合取项, 整体当叶子返回 —— 它们给不出索引边界,
+// 但必须原样保留下来做行级过滤。
+func flattenAnd(cond Condition, out *[]Condition) {
+	if and, ok := cond.(AndCondition); ok {
+		for _, sub := range and.Conditions {
+			flattenAnd(sub, out)
 		}
+		return
+	}
+	*out = append(*out, cond)
+}
 
-		if hasStart || hasEnd {
-			var rem Condition
-			if len(remaining) == 1 {
-				rem = remaining[0]
-			} else if len(remaining) > 1 {
-				rem = And(remaining...)
-			}
-			return sk, ek, si, ei, rem
+// incBE 把 buf 当作大端无符号整数加一 (原地)。返回 false 表示溢出 ——
+// 此时 buf 之后不再存在任何更大的键。
+func incBE(buf []byte) bool {
+	for i := len(buf) - 1; i >= 0; i-- {
+		buf[i]++
+		if buf[i] != 0 {
+			return true
 		}
 	}
-	return nil, nil, false, false, cond
+	return false
+}
+
+// fillFF 把 buf[from:] 全置 0xFF (该分量及其后分量的最大值)。
+// from 可以是 0 —— 首分量就是范围分量时前缀为空, 此时整个键都无上界。
+func fillFF(buf []byte, from int) {
+	for i := from; i < len(buf); i++ {
+		buf[i] = 0xFF
+	}
 }
 
 // buildScanPlan 分析 WHERE 条件, 选择最优索引
@@ -183,8 +335,11 @@ func (q *Query) buildScanPlan() *scanPlan {
 		return nil
 	}
 
-	var bestPlan *scanPlan
+	// 先展平成合取项: 索引区间是按"分量"折叠出来的, 只有 AND 语义才可加。
+	var leaves []Condition
+	flattenAnd(q.whereCond, &leaves)
 
+	var bestPlan *scanPlan
 	for i := range q.table.indexMeta.metas {
 		m := &q.table.indexMeta.metas[i]
 		if m.Flag&0x80 != 0 {
@@ -194,34 +349,141 @@ func (q *Query) buildScanPlan() *scanPlan {
 		if !ok {
 			continue
 		}
+		plan := q.planIndex(m, handler, leaves)
+		if plan == nil {
+			continue // 此索引不匹配
+		}
 		// 该索引是否承载行数据: 由处理器自报, 本层不认识索引类型
-		rowStore := handler.IsRowStore()
-		fieldName := q.fieldNameFromMeta(m)
-		ft := q.fieldTypeFromMeta(m)
-		if fieldName == "" {
-			continue
+		plan.rowStore = handler.IsRowStore()
+		if plan.rowStore {
+			return plan // 行存储索引最优 (免回表), 不再找
 		}
-
-		sk, ek, si, ei, rem := q.condToRange(q.whereCond, fieldName, ft)
-		if sk == nil && ek == nil {
-			continue // 此索引不匹配
+		if bestPlan == nil {
+			bestPlan = plan
 		}
+	}
+	return bestPlan
+}
+
+// planIndex 为一个索引推导扫描区间; 该索引用不上时返回 nil。
+func (q *Query) planIndex(m *IndexMeta, handler index.IndexHandler, leaves []Condition) *scanPlan {
+	lay, ok := q.buildKeyLayout(m)
+	if !ok {
+		return nil
+	}
+	n := len(lay.sizes)
 
-		plan := &scanPlan{
-			meta: m, handler: handler,
-			startKey: sk, endKey: ek,
-			startInclusive: si, endInclusive: ei,
-			rowStore: rowStore, remainingCond: rem,
+	cons := make([]keyConstraint, n)   // 各分量合并后的边界
+	compCond := make([][]Condition, n) // 各分量上的原始条件 (用于决定哪些能丢)
+	var rest []Condition               // 与索引无关、必须保留的条件
+
+	for _, c := range leaves {
+		k, value, kind := q.classifyLeaf(c, m)
+		if k < 0 {
+			rest = append(rest, c)
+			continue
 		}
-		// 优先选承载行数据的索引 (避免回查)
-		if rowStore || bestPlan == nil {
-			bestPlan = plan
+		ft := q.table.Field[m.FieldIDs[k]].FieldType
+		// 用 Safe 版本: 条件的值类型与列存储类型不符时 encodeIndexKey 会 panic,
+		// 规划阶段不该把用户输入直接变成崩溃 —— 退回行级过滤由 Match 去处理。
+		enc, err := encodeIndexKeySafe(value, ft)
+		if err != nil || len(enc) != lay.sizes[k] {
+			// 编码失败, 或长度与分量长度不符 (拼进定长键会错位) → 交给行级过滤
+			rest = append(rest, c)
+			continue
 		}
-		if rowStore {
-			break // 行存储索引最优, 不再找
+		if !cons[k].add(kind, enc) {
+			// 该分量上的条件互相矛盾 → 结果必为空。
+			// 这里保守地"放弃索引"而不是返回空区间: 定长补码编码下负数字节序与
+			// 数值序相反, 用字节比较断定"空"对负数不成立, 退回全表扫描更稳。
+			return nil
 		}
+		compCond[k] = append(compCond[k], c)
+	}
+
+	if cons[0].count == 0 {
+		return nil // 首分量没有边界 → 索引给不出连续区间
+	}
+
+	// 从左往右消费"等值分量"; 之后最多再消费一个范围分量, 更后面的分量帮不上忙。
+	eqLen := 0
+	for eqLen < n && cons[eqLen].isEquality() {
+		eqLen++
+	}
+	rangeIdx := -1
+	if eqLen < n && cons[eqLen].count > 0 {
+		rangeIdx = eqLen
+	}
+
+	startKey := make([]byte, lay.total)
+	endKey := make([]byte, lay.total)
+	prefix := 0
+	for j := 0; j < eqLen; j++ {
+		copy(startKey[prefix:], cons[j].lo)
+		copy(endKey[prefix:], cons[j].lo)
+		prefix += len(cons[j].lo)
+	}
+
+	plan := &scanPlan{
+		meta: m, handler: handler,
+		startKey: startKey, endKey: endKey,
+		startInclusive: true, endInclusive: true,
+		remainingCond: q.remainCond(rest, compCond, eqLen, rangeIdx),
+	}
+
+	// 全部等值 → 精确键 (此时 prefix == total, fillFF 无事可做);
+	// 只消费了等值前缀 → 该前缀下的全部键: [前缀||00.., 前缀||FF..]。
+	if eqLen == n || rangeIdx < 0 {
+		fillFF(endKey, prefix)
+		return plan
+	}
+
+	rc := &cons[rangeIdx]
+	size := lay.sizes[rangeIdx]
+
+	// 下界: [前缀 || 分量值 || 00..]
+	if rc.lo != nil {
+		copy(startKey[prefix:], rc.lo)
+		if !rc.loInc {
+			// 严格下界: 要跳过分量 == lo 的全部键, 即把 (前缀||lo) 整体加一
+			if !incBE(startKey[:prefix+size]) {
+				return nil // 已经是最大键, 该条件不可能有匹配行
+			}
+		}
+	}
+
+	// 上界: [前缀 || 分量值 || FF..]
+	if rc.hi == nil {
+		fillFF(endKey, prefix)
+	} else if rc.hiInc {
+		copy(endKey[prefix:], rc.hi)
+		fillFF(endKey, prefix+size)
+	} else {
+		// 严格上界: 分量 == hi 的键全部排除
+		copy(endKey[prefix:], rc.hi)
+		plan.endInclusive = false
+	}
+	return plan
+}
+
+// remainCond 汇总索引区间没能表达、仍需逐行过滤的条件。
+// 已被区间精确覆盖的分量 (等值前缀 + 那个范围分量) 上的条件可以安全丢掉。
+func (q *Query) remainCond(rest []Condition, compCond [][]Condition, eqLen, rangeIdx int) Condition {
+	remaining := rest
+	for j := range compCond {
+		if j < eqLen || j == rangeIdx {
+			continue
+		}
+		remaining = append(remaining, compCond[j]...)
+	}
+	switch len(remaining) {
+	case 0:
+		return nil
+	case 1:
+		return remaining[0]
+	default:
+		return And(remaining...)
 	}
-	return bestPlan
 }
 
 // ======================== 核心查询 ========================

DB/core/zz_composite_index_test.go

@@ -0,0 +1,295 @@
+package core
+
+import (
+	"os"
+	"sort"
+	"testing"
+	"unsafe"
+
+	"TsDB2/DB/core/page"
+	"TsDB2/DB/core/types"
+)
+
+// 本文件锁定复合索引 (多分量索引) 的端到端行为。
+//
+// 索引键 = 各分量字段定长拼接。修复前 extractIndexKey 只取 FieldIDs[0], 查询侧也只按
+// 首分量推导区间 —— 于是 CreateIndex 收下多个字段, 实际建出的却是"只含首列"的索引:
+// 查询结果仍然正确 (残余条件后置过滤兜住了), 但索引形同虚设, 且页里的键只有 8 字节,
+// 本该 16 字节的复合点查根本命不中。
+//
+// 所以下面第一条断言直接看索引页的 EntryLen —— 它不经过查询路径, 是最硬的证据。
+
+const compPath = "zz_composite_index.db"
+
+// compKey 走公开 API 构造复合索引键
+func compKey(t *testing.T, tbl *Table, values ...interface{}) []byte {
+	t.Helper()
+	k, err := tbl.BuildIndexKey("idx_ab", values...)
+	if err != nil {
+		t.Fatalf("BuildIndexKey: %v", err)
+	}
+	return k
+}
+
+// compGet 从一行结果里取出某列的值 (测试表全是 8 字节列)
+func compGet(t *testing.T, row []types.FieldValue, col string) uint64 {
+	t.Helper()
+	for _, f := range row {
+		if string(f.FieldName) != col {
+			continue
+		}
+		switch v := f.Value.(type) {
+		case uint64:
+			return v
+		case int64:
+			return uint64(v)
+		case uint32:
+			return uint64(v)
+		case int32:
+			return uint64(int64(v))
+		}
+		t.Fatalf("列 %s 的值类型意外: %T", col, f.Value)
+	}
+	t.Fatalf("结果里没有列 %s", col)
+	return 0
+}
+
+// compCs 取出结果集里所有 C 列的值并排序, 便于与期望集合直接比对
+func compCs(t *testing.T, rows [][]types.FieldValue) []uint64 {
+	t.Helper()
+	got := make([]uint64, 0, len(rows))
+	for _, r := range rows {
+		got = append(got, compGet(t, r, "C"))
+	}
+	sort.Slice(got, func(i, j int) bool { return got[i] < got[j] })
+	return got
+}
+
+func TestCompositeIndexEndToEnd(t *testing.T) {
+	os.Remove(compPath)
+	os.Remove(compPath + ".wal")
+	t.Cleanup(func() {
+		if t.Failed() {
+			return // 失败时保留测试库, 便于排查
+		}
+		_ = os.Remove(compPath)
+		_ = os.Remove(compPath + ".wal")
+	})
+
+	db, err := Open(compPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db.SetAutoCommitDelay(0)
+
+	tbl, err := db.CreateTable("T", []types.FieldInfo{
+		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
+		{FieldType: types.FieldType_8byte, FieldName: []byte("A")},
+		{FieldType: types.FieldType_8byte, FieldName: []byte("B")},
+		{FieldType: types.FieldType_8byte, FieldName: []byte("C")},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tbl.CreateIndex("pk_id", page.IndexTypePrimary, []byte{0}); err != nil {
+		t.Fatal(err)
+	}
+	// 复合索引 (A, B): 分量顺序即 fieldIDs 顺序
+	if _, err := tbl.CreateIndex("idx_ab", page.IndexTypeBTree, []byte{1, 2}); err != nil {
+		t.Fatal(err)
+	}
+
+	// 5x5 行: ID = a*5+b+1, A = a, B = b, C = a*100+b
+	aVal := func(id uint64) uint64 { return (id - 1) / 5 }
+	bVal := func(id uint64) uint64 { return (id - 1) % 5 }
+	for id := uint64(1); id <= 25; id++ {
+		if _, err := tbl.InsertRow([]types.FieldValue{
+			{FieldName: []byte("ID"), Value: id},
+			{FieldName: []byte("A"), Value: aVal(id)},
+			{FieldName: []byte("B"), Value: bVal(id)},
+			{FieldName: []byte("C"), Value: aVal(id)*100 + bVal(id)},
+		}); err != nil {
+			t.Fatalf("insert %d: %v", id, err)
+		}
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	// 期望行数 25, 每条键 8B 主键 + 16B 复合键
+	if n, err := tbl.Query().Count(); err != nil || n != 25 {
+		t.Fatalf("初始行数 = %d (err=%v), 期望 25", n, err)
+	}
+
+	// ===== 断言 1 (最硬): 索引页里每条记录的键是 16 字节, 而不是 8 字节 =====
+	meta := rebuildIndexMetaByName(t, tbl, "idx_ab")
+	buf := rebuildReadPage(t, tbl, meta.EntryPage)
+	hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
+	wantEntry := 8 + 2*8
+	if int(hdr.EntryLen) != wantEntry {
+		t.Fatalf("idx_ab 条目长度 = %d, 期望 %d (= 8B 主键 + 16B 复合键); "+
+			"键只有 8 字节, 说明索引退化成了[只含首列]的索引 (EntryPage=%d)",
+			hdr.EntryLen, wantEntry, meta.EntryPage)
+	}
+	if hdr.IsLeaf == 0 {
+		t.Fatalf("25 行时 idx_ab 根页应为叶子, EntryPage=%d", meta.EntryPage)
+	}
+	if int(hdr.Count) != 25 {
+		t.Fatalf("idx_ab 条目数 = %d, 期望 25", hdr.Count)
+	}
+
+	// ===== 断言 2: 复合点查 (公开 API) =====
+	k23 := compKey(t, tbl, uint64(2), uint64(3))
+	if len(k23) != 16 {
+		t.Fatalf("复合键长度 = %d, 期望 16", len(k23))
+	}
+	row, err := tbl.SearchByIndex("idx_ab", k23)
+	if err != nil || row == nil {
+		t.Fatalf("复合点查 (A=2,B=3) 失败: %v", err)
+	}
+	if got := compGet(t, row, "C"); got != 203 {
+		t.Fatalf("复合点查命中 C=%d, 期望 203", got)
+	}
+
+	// ===== 断言 3: 不存在的复合键必须落空 =====
+	//
+	// 注意 B+ 树的键比较取两者较短长度 (sbtCmp 用 min(len)), 所以只给前缀分量
+	// (be8(2)) 时按前缀匹配、会落到 A=2 的第一条 —— 这是既有的前缀语义, 不是缺陷。
+	// 要做完整定位就得给足 16 字节, 这一点由断言 2 的 C==203 保证。
+	// 这里验证的是"全键但值不存在"时不能误命中邻近条目。
+	if r, _ := tbl.SearchByIndex("idx_ab", compKey(t, tbl, uint64(2), uint64(9))); r != nil {
+		t.Fatalf("不存在的复合键 (A=2,B=9) 命中了 C=%d", compGet(t, r, "C"))
+	}
+
+	// ===== 断言 4: 各种区间组合的正确性 =====
+	assertC := func(name string, cond Condition, want ...uint64) {
+		t.Helper()
+		rows, err := tbl.Query().Where(cond).Find()
+		if err != nil {
+			t.Fatalf("%s: %v", name, err)
+		}
+		got := compCs(t, rows)
+		wantSorted := append([]uint64(nil), want...)
+		sort.Slice(wantSorted, func(i, j int) bool { return wantSorted[i] < wantSorted[j] })
+		if len(got) != len(wantSorted) {
+			t.Fatalf("%s: 命中 %v (%d 行), 期望 %v (%d 行)", name, got, len(got), wantSorted, len(wantSorted))
+		}
+		for i := range wantSorted {
+			if got[i] != wantSorted[i] {
+				t.Fatalf("%s: 命中 %v, 期望 %v", name, got, wantSorted)
+			}
+		}
+	}
+	eq8 := func(col string, v uint64) Condition { return Eq(col, v, types.FieldType_8byte) }
+	ge8 := func(col string, v uint64) Condition { return Ge(col, v, types.FieldType_8byte) }
+	gt8 := func(col string, v uint64) Condition { return Gt(col, v, types.FieldType_8byte) }
+	le8 := func(col string, v uint64) Condition { return Le(col, v, types.FieldType_8byte) }
+	lt8 := func(col string, v uint64) Condition { return Lt(col, v, types.FieldType_8byte) }
+
+	assertC("全键等值 A=2,B=3", And(eq8("A", 2), eq8("B", 3)), 203)
+	assertC("最左前缀 A=2", eq8("A", 2), 200, 201, 202, 203, 204)
+	assertC("前缀+次分量 >= B>=2", And(eq8("A", 2), ge8("B", 2)), 202, 203, 204)
+	assertC("前缀+次分量 <= B<=1", And(eq8("A", 2), le8("B", 1)), 200, 201)
+	assertC("前缀+严格区间 1<B<3", And(eq8("A", 2), gt8("B", 1), lt8("B", 3)), 202)
+	assertC("首分量范围 A>=3", ge8("A", 3),
+		300, 301, 302, 303, 304, 400, 401, 402, 403, 404)
+	assertC("首分量严格上界 A<1", lt8("A", 1), 0, 1, 2, 3, 4)
+	// 只约束尾分量: 首分量无边界的索引给不出区间, 必须退回全表扫描且结果仍正确
+	assertC("仅尾分量 B=4", eq8("B", 4), 4, 104, 204, 304, 404)
+	// 非等值/非范围条件不能进索引, 只能后置过滤
+	assertC("前缀+NOT 混合", And(eq8("A", 2), Not(eq8("B", 3))), 200, 201, 202, 204)
+
+	// ===== 断言 5: 更新要同时搬走旧的复合条目 =====
+	if err := tbl.UpdateRow("idx_ab", k23, []types.FieldValue{
+		{FieldName: []byte("B"), Value: uint64(7)},
+	}); err != nil {
+		t.Fatalf("UpdateRow: %v", err)
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if r, _ := tbl.SearchByIndex("idx_ab", k23); r != nil {
+		t.Fatalf("更新后旧复合键 (A=2,B=3) 仍能命中: C=%d", compGet(t, r, "C"))
+	}
+	k27 := compKey(t, tbl, uint64(2), uint64(7))
+	row27, err := tbl.SearchByIndex("idx_ab", k27)
+	if err != nil || row27 == nil {
+		t.Fatalf("更新后新复合键 (A=2,B=7) 查不到: %v", err)
+	}
+	if got := compGet(t, row27, "C"); got != 203 {
+		t.Fatalf("更新后 C=%d, 期望 203 (未改动的列应保持)", got)
+	}
+	assertC("更新后 A=2,B=7", And(eq8("A", 2), eq8("B", 7)), 203)
+	assertC("更新后旧值 A=2,B=3 查不到", And(eq8("A", 2), eq8("B", 3)))
+
+	// ===== 断言 6: 删除 =====
+	if err := tbl.DeleteRow("idx_ab", k27); err != nil {
+		t.Fatalf("DeleteRow: %v", err)
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if r, _ := tbl.SearchByIndex("idx_ab", k27); r != nil {
+		t.Fatalf("删除后复合键仍能命中: C=%d", compGet(t, r, "C"))
+	}
+	if n, err := tbl.Query().Count(); err != nil || n != 24 {
+		t.Fatalf("删除后行数 = %d (err=%v), 期望 24", n, err)
+	}
+
+	// ===== 断言 7: 重启后复合索引仍可用 (键在位、能读回) =====
+	if err := db.Close(); err != nil {
+		t.Fatal(err)
+	}
+	db, err = Open(compPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db.SetAutoCommitDelay(0)
+	tbl, ok := db.FindTable("T")
+	if !ok {
+		t.Fatal("重启后找不到表 T")
+	}
+	if r, err := tbl.SearchByIndex("idx_ab", compKey(t, tbl, uint64(3), uint64(4))); err != nil || r == nil {
+		t.Fatalf("重启后复合点查 (A=3,B=4) 失败: %v", err)
+	}
+	assertC("重启后 A=1,B=1", And(eq8("A", 1), eq8("B", 1)), 101)
+
+	// ===== 断言 8: 批量删除的重建路径用的是同一套键编码 =====
+	del, err := tbl.Delete().Where(eq8("B", 0)).Exec()
+	if err != nil {
+		t.Fatalf("批量删除: %v", err)
+	}
+	if del != 5 {
+		t.Fatalf("批量删除行数 = %d, 期望 5", del)
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if r, err := tbl.SearchByIndex("idx_ab", compKey(t, tbl, uint64(3), uint64(1))); err != nil || r == nil {
+		t.Fatalf("重建后复合点查 (A=3,B=1) 失败: %v", err)
+	}
+	if r, _ := tbl.SearchByIndex("idx_ab", compKey(t, tbl, uint64(3), uint64(0))); r != nil {
+		t.Fatalf("重建后残留已删行 (A=3,B=0): C=%d", compGet(t, r, "C"))
+	}
+	if n, err := tbl.Query().Count(); err != nil || n != 19 {
+		t.Fatalf("批量删除后行数 = %d (err=%v), 期望 19", n, err)
+	}
+	assertC("重建后前缀 A=4", eq8("A", 4), 401, 402, 403, 404)
+
+	// ===== 断言 9: 复合索引的构造约束 =====
+	if _, err := tbl.CreateIndex("idx_dup", page.IndexTypeBTree, []byte{1, 1}); err == nil {
+		t.Fatal("重复列 (1,1) 建索引应当报错, 否则索引会永远取不齐分量而静默为空")
+	}
+	if _, err := tbl.CreateIndex("idx_pk2", page.IndexTypePrimary, []byte{0, 1}); err == nil {
+		t.Fatal("行存储索引 (主键) 的复合键应当被拒绝")
+	}
+	if _, err := tbl.BuildIndexKey("idx_ab", uint64(1)); err == nil {
+		t.Fatal("分量个数不足时 BuildIndexKey 应当报错")
+	}
+	if _, err := tbl.BuildIndexKey("idx_ab", uint64(1), "x"); err == nil {
+		t.Fatal("分量值类型不符时 BuildIndexKey 应当报错而不是 panic")
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+}

DB/db.go

@@ -163,25 +163,33 @@ func UseTable[T any](db *DB, name ...string) (*Table[T], error) {
 		if err != nil {
 			return nil, fmt.Errorf("vrtx: create index for %s: %w", tableName, err)
 		}
-		// 自动创建标签声明的二级索引
+		// 自动创建标签声明的二级索引 (同名字段已合并为复合索引)
 		for _, idx := range model.Indexes {
-			name := idx.Name
-			if name == "" {
-				name = "idx_" + tableName + "_" + idx.ColName
+			name := idx.indexName(tableName)
+			if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs()); err != nil {
+				return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
 			}
-			_, _ = coreTbl.CreateIndex(name, idx.IndexType, []byte{idx.FieldID})
 		}
 	} else {
 		// 表已存在，自动同步 schema
 		syncSchema(coreTbl, model)
 		// 检查并补充缺失的二级索引
 		for _, idx := range model.Indexes {
-			name := idx.Name
-			if name == "" {
-				name = "idx_" + tableName + "_" + idx.ColName
+			name := idx.indexName(tableName)
+			m, found := coreTbl.IndexMeta().FindIndex(name)
+			if !found {
+				if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs()); err != nil {
+					return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
+				}
+				continue
 			}
-			if _, found := coreTbl.IndexMeta().FindIndex(name); !found {
-				_, _ = coreTbl.CreateIndex(name, idx.IndexType, []byte{idx.FieldID})
+			// 已存在: 校验分量列表与模型一致。
+			// 不一致说明索引是按旧结构建的 (例如后来才把同名字段改成复合索引),
+			// 静默沿用会一直用一个"少一列"的索引 —— 查询结果对但索引形同虚设,
+			// 这里直接报错, 让人决定是重建还是改名。
+			if !sameIndexFields(m, idx.coreFieldIDs()) {
+				return nil, fmt.Errorf("vrtx: 索引 %s 已存在但分量列表不同 (已有 %d 列, 模型期望 %d 列), 请先删除该索引",
+					name, m.FieldCnt, len(idx.coreFieldIDs()))
 			}
 		}
 	}
@@ -195,6 +203,20 @@ func UseTable[T any](db *DB, name ...string) (*Table[T], error) {
 	return tbl, nil
 }
 
+// sameIndexFields 比较已有索引的分量列表与模型期望是否一致 (顺序敏感)
+func sameIndexFields(m *core.IndexMeta, want []byte) bool {
+	n := int(m.FieldCnt)
+	if n != len(want) {
+		return false
+	}
+	for i := 0; i < n; i++ {
+		if m.FieldIDs[i] != want[i] {
+			return false
+		}
+	}
+	return true
+}
+
 // syncSchema 将 DB schema 同步为与 struct 一致
 // struct 多 → AddField；struct 少 → DropField
 func syncSchema(coreTbl *core.Table, model *ModelInfo) {

DB/types.go

@@ -20,13 +20,30 @@ type FieldDef struct {
 
 // IndexDef 从结构体标签解析的二级索引定义
 type IndexDef struct {
-	Name      string         // 索引名 (空=自动生成 idx_表名_字段名)
-	FieldID   byte           // 字段在 CoreFields 中的位置
-	ColName   string         // 字段列名 (用于自动生成索引名)
+	Name      string         // 索引名 (空=自动生成 idx_表名_首个字段名)
+	FieldID   byte           // 首个字段在 CoreFields 中的位置 (单列索引)
+	FieldIDs  []byte         // 索引分量在 CoreFields 中的位置, 按键序 (复合索引)
+	ColName   string         // 首个字段列名 (用于自动生成索引名)
 	IndexType page.IndexType // 索引类型 (默认 BTree)
 	Unique    bool           // 唯一索引 (预留)
 }
 
+// coreFieldIDs 返回 core.CreateIndex 需要的分量列表
+func (d IndexDef) coreFieldIDs() []byte {
+	if len(d.FieldIDs) > 0 {
+		return d.FieldIDs
+	}
+	return []byte{d.FieldID}
+}
+
+// indexName 返回索引的最终名字: 显式命名优先, 否则 idx_表名_首个字段名
+func (d IndexDef) indexName(tableName string) string {
+	if d.Name != "" {
+		return d.Name
+	}
+	return "idx_" + tableName + "_" + d.ColName
+}
+
 // ModelInfo 模型元信息
 type ModelInfo struct {
 	Fields       []FieldDef        // 用户定义字段 (不含自动主键)
@@ -171,8 +188,14 @@ func ParseModel(obj interface{}) (*ModelInfo, error) {
 	}
 	info.CorePKField = byte(info.PKIndex)
 
-	// 构建 Indexes: 收集所有标记了 index 的字段的索引定义
+	// 构建 Indexes: 收集所有标记了 index 的字段的索引定义。
+	//
+	// 同一个具名索引 (index:NAME) 被多个字段声明时, 合并成一个复合索引:
+	// 分量顺序 = 字段在表中的物理顺序。这样无需在 tag 里额外表达键顺序,
+	// 也正好对上复合索引"最左前缀"的用法 (a=? 或 a=? AND b>?)。
+	// 未命名的 index 仍各自成单列索引。同名时以第一个声明为准 (含索引类型)。
 	info.Indexes = nil
+	named := make(map[string]int, len(info.AllFields))
 	for i, f := range info.AllFields {
 		if i == info.PKIndex {
 			continue // 跳过主键
@@ -183,13 +206,24 @@ func ParseModel(obj interface{}) (*ModelInfo, error) {
 			continue
 		}
 		_, idxDef, err := parseField(*sf, f.Index)
-		if err != nil {
+		if err != nil || idxDef == nil {
 			continue
 		}
-		if idxDef != nil {
-			idxDef.FieldID = byte(i)
-			idxDef.ColName = f.ColName
-			info.Indexes = append(info.Indexes, *idxDef)
+		if idxDef.Name != "" {
+			if pos, ok := named[idxDef.Name]; ok {
+				// 同名 → 追加为新的索引分量
+				info.Indexes[pos].FieldIDs = append(info.Indexes[pos].FieldIDs, byte(i))
+				info.Indexes[pos].Unique = info.Indexes[pos].Unique || idxDef.Unique
+				continue
+			}
+		}
+		d := *idxDef
+		d.FieldID = byte(i)
+		d.FieldIDs = []byte{byte(i)}
+		d.ColName = f.ColName
+		info.Indexes = append(info.Indexes, d)
+		if d.Name != "" {
+			named[d.Name] = len(info.Indexes) - 1
 		}
 	}
 

DB/types_index_test.go

@@ -0,0 +1,173 @@
+package DB
+
+import (
+	"os"
+	"sort"
+	"testing"
+)
+
+// 同名索引标签 → 复合索引:
+//   - Dept 与 Age 都声明 index:idx_dept_age → 合成一个 2 分量索引;
+//   - Score 单独一个具名索引;
+//   - Level 未命名 → 仍按单列索引自动命名。
+type compEmp struct {
+	BaseModel
+	Name  string `vrtx:"maxlen:32"`
+	Dept  string `vrtx:"maxlen:16,index:idx_dept_age"`
+	Age   int64  `vrtx:"index:idx_dept_age"`
+	Score int64  `vrtx:"index:idx_score"`
+	Level int64  `vrtx:"index"`
+}
+
+// compEmpOneField 列完全相同, 只是 idx_dept_age 少了一个字段 —— 用来触发
+// "已有索引分量数与模型不符" 的校验。列集合一致, 所以不会牵扯到 schema 同步。
+type compEmpOneField struct {
+	BaseModel
+	Name  string `vrtx:"maxlen:32"`
+	Dept  string `vrtx:"maxlen:16,index:idx_dept_age"`
+	Age   int64
+	Score int64
+	Level int64
+}
+
+func TestParseModelMergesSameNameIndex(t *testing.T) {
+	m, err := ParseModel(&compEmp{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	byName := map[string]IndexDef{}
+	unnamed := 0
+	for _, d := range m.Indexes {
+		if d.Name == "" {
+			unnamed++
+			continue
+		}
+		byName[d.Name] = d
+	}
+
+	comp, ok := byName["idx_dept_age"]
+	if !ok {
+		t.Fatalf("没解析出 idx_dept_age, 实际索引: %+v", m.Indexes)
+	}
+	// 列顺序: 0=_id, 1=Name, 2=Dept, 3=Age; 分量顺序 = 字段物理顺序
+	if got := comp.coreFieldIDs(); len(got) != 2 || got[0] != 2 || got[1] != 3 {
+		t.Fatalf("idx_dept_age 分量 = %v, 期望 [2 3] (Dept, Age)", got)
+	}
+	if comp.ColName != "Dept" {
+		t.Fatalf("idx_dept_age 首列名 = %q, 期望 Dept", comp.ColName)
+	}
+
+	if d, ok := byName["idx_score"]; !ok || len(d.coreFieldIDs()) != 1 {
+		t.Fatalf("idx_score 应当是单列索引, 实际 %+v", d)
+	}
+	if unnamed != 1 {
+		t.Fatalf("未命名索引数 = %d, 期望 1 (Level)", unnamed)
+	}
+	// 未命名索引的自动命名: idx_表名_列名
+	for _, d := range m.Indexes {
+		if d.Name == "" {
+			if got := d.indexName("t"); got != "idx_t_Level" {
+				t.Fatalf("未命名索引自动命名 = %q, 期望 %q", got, "idx_t_Level")
+			}
+		}
+	}
+}
+
+func TestORMCompositeIndexEndToEnd(t *testing.T) {
+	path := "zz_orm_composite.db"
+	os.Remove(path)
+	os.Remove(path + ".wal")
+	t.Cleanup(func() {
+		if t.Failed() {
+			return
+		}
+		_ = os.Remove(path)
+		_ = os.Remove(path + ".wal")
+	})
+
+	db, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	db.SetAutoCommitDelay(0)
+
+	tbl, err := UseTable[compEmp](db, "comp_emp")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// 核心层应当真的建出了 2 分量索引
+	meta, found := tbl.coreTable.IndexMeta().FindIndex("idx_dept_age")
+	if !found {
+		t.Fatal("core 层没有 idx_dept_age")
+	}
+	if meta.FieldCnt != 2 {
+		t.Fatalf("idx_dept_age 分量数 = %d, 期望 2", meta.FieldCnt)
+	}
+
+	// Dept 只有 d0/d1, Age 在 0..4 之间循环, Score = i*10
+	depts := []string{"d0", "d1"}
+	for i := int64(0); i < 20; i++ {
+		if err := tbl.Insert(&compEmp{
+			Name: "n", Dept: depts[i%2], Age: i % 5, Score: i * 10, Level: i % 3,
+		}); err != nil {
+			t.Fatalf("insert %d: %v", i, err)
+		}
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	scores := func(q *Query[compEmp]) []int64 {
+		t.Helper()
+		rows, err := q.Find()
+		if err != nil {
+			t.Fatalf("find: %v", err)
+		}
+		out := make([]int64, 0, len(rows))
+		for _, r := range rows {
+			out = append(out, r.Score)
+		}
+		sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
+		return out
+	}
+	eq := func(got, want []int64) bool {
+		if len(got) != len(want) {
+			return false
+		}
+		for i := range got {
+			if got[i] != want[i] {
+				return false
+			}
+		}
+		return true
+	}
+
+	// 最左前缀: Dept = d0 (i 为偶数)
+	if got := scores(tbl.Where("Dept", "=", "d0")); !eq(got, []int64{0, 20, 40, 60, 80, 100, 120, 140, 160, 180}) {
+		t.Fatalf("Dept=d0 命中 Score=%v", got)
+	}
+	// 全键等值: Dept = d0 AND Age = 3 → i ∈ {8, 18}
+	if got := scores(tbl.Where("Dept", "=", "d0").Where("Age", "=", int64(3))); !eq(got, []int64{80, 180}) {
+		t.Fatalf("Dept=d0 AND Age=3 命中 Score=%v, 期望 [80 180]", got)
+	}
+	// 前缀等值 + 次分量范围: Dept = d0 AND Age > 2 → i ∈ {4, 8, 14, 18}
+	if got := scores(tbl.Where("Dept", "=", "d0").Where("Age", ">", int64(2))); !eq(got, []int64{40, 80, 140, 180}) {
+		t.Fatalf("Dept=d0 AND Age>2 命中 Score=%v, 期望 [40 80 140 180]", got)
+	}
+	// 只约束尾分量: 索引帮不上忙, 结果仍要对
+	if got := scores(tbl.Where("Age", "=", int64(4))); !eq(got, []int64{40, 90, 140, 190}) {
+		t.Fatalf("Age=4 命中 Score=%v, 期望 [40 90 140 190]", got)
+	}
+
+	// 重复 UseTable: 分量一致, 不能误报
+	if _, err := UseTable[compEmp](db, "comp_emp"); err != nil {
+		t.Fatalf("重复 UseTable 应当通过: %v", err)
+	}
+	// 把同名字段改成单列索引 → 与库里的 2 分量索引不符, 必须报错而不是静默沿用
+	if _, err := UseTable[compEmpOneField](db, "comp_emp"); err == nil {
+		t.Fatal("idx_dept_age 分量数不符时 UseTable 应当报错, 否则会一直用一个少一列的索引")
+	}
+}