TsDB2/DB/query.go

package DB

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"fmt"
	"reflect"
)

// Query 泛型查询构建器
type Query[T any] struct {
	table      *Table[T]
	wheres     []whereClause
	orders     []orderClause
	selectCols []string
	limit      int
	offset     int
}

type whereClause struct {
	field string
	op    string
	value interface{}
}

type orderClause struct {
	field string
	desc  bool
}

func (q *Query[T]) addWhere(field, op string, value interface{}) *Query[T] {
	q.wheres = append(q.wheres, whereClause{field: field, op: op, value: value})
	return q
}

// Where 添加条件
func (q *Query[T]) Where(field string, op string, value interface{}) *Query[T] {
	return q.addWhere(field, op, value)
}

// OrderBy 排序
func (q *Query[T]) OrderBy(field string, desc ...bool) *Query[T] {
	d := false
	if len(desc) > 0 && desc[0] {
		d = true
	}
	q.orders = append(q.orders, orderClause{field: field, desc: d})
	return q
}

// Limit 限制行数
func (q *Query[T]) Limit(n int) *Query[T] {
	q.limit = n
	return q
}

// Offset 跳过行数
func (q *Query[T]) Offset(n int) *Query[T] {
	q.offset = n
	return q
}

// Select 只取指定列 (Go 字段名)。投影下推到引擎: 未选中的列不做反序列化与
// 堆页解引用; 读出的对象里未选中的字段保持零值。
func (q *Query[T]) Select(cols ...string) *Query[T] {
	q.selectCols = cols
	return q
}

// Find 执行查询返回所有结果
//
// 用 ScanWithPKID 而不是 Find: 行号不在字段值里, 但它决定了返回的对象能不能被精确
// 地改 / 删 —— 同一个时间戳下有多行时, 只有行号指得清"返回的是哪一行"。
func (q *Query[T]) Find() ([]*T, error) {
	cq := q.buildCoreQuery()
	result := make([]*T, 0)
	var convErr error
	err := cq.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		obj, err := q.table.convertRow(pkID, row)
		if err != nil {
			// 反序列化失败: 停止扫描并让 Find 返回错误, 不能静默跳过该行
			if convErr == nil {
				convErr = err
			}
			return false
		}
		result = append(result, obj)
		return true
	})
	if err != nil {
		return nil, err
	}
	if convErr != nil {
		return nil, convErr
	}
	return result, nil
}

// First 返回第一条结果
func (q *Query[T]) First() (*T, error) {
	q.limit = 1
	results, err := q.Find()
	if err != nil {
		return nil, err
	}
	if len(results) == 0 {
		return nil, fmt.Errorf("tsdb: record not found")
	}
	return results[0], nil
}

// Count 返回匹配行数
func (q *Query[T]) Count() (int64, error) {
	cq := q.buildCoreQuery()
	return cq.Count()
}

// Scan 流式遍历
func (q *Query[T]) Scan(callback func(*T) bool) error {
	cq := q.buildCoreQuery()
	var convErr error
	err := cq.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		obj, err := q.table.convertRow(pkID, row)
		if err != nil {
			// 反序列化失败: 停止扫描并让 Scan 返回错误, 不能静默跳过该行
			if convErr == nil {
				convErr = err
			}
			return false
		}
		return callback(obj)
	})
	if err != nil {
		return err
	}
	return convErr
}

// buildCoreQuery 构建 core 层的 Query
func (q *Query[T]) buildCoreQuery() *core.Query {
	cq := q.table.coreTable.Query()

	// 构建 WHERE 条件
	if len(q.wheres) > 0 {
		cond := q.buildCondition(q.wheres[0])
		for i := 1; i < len(q.wheres); i++ {
			cond = core.And(cond, q.buildCondition(q.wheres[i]))
		}
		cq = cq.Where(cond)
	}

	// ORDER BY
	for _, o := range q.orders {
		cq = cq.OrderBy(o.field, o.desc)
	}

	// SELECT 投影下推
	if len(q.selectCols) > 0 {
		cq = cq.Select(q.selectCols...)
	}

	// LIMIT / OFFSET
	if q.limit > 0 {
		cq = cq.Limit(q.limit)
	}
	if q.offset > 0 {
		cq = cq.Offset(q.offset)
	}

	return cq
}

// buildCondition 将 whereClause 转换为 core.Condition
func (q *Query[T]) buildCondition(w whereClause) core.Condition {
	// ft 只用来把条件值归一化成列期望的 Go 类型 (normalizeCondValue);
	// 条件构造函数的第三个参数已废弃 —— 比较一律按列的实际存储类型进行
	// (见 core.Table.compareTypeFor), 故统一传 types.FieldType(0)。
	ft := q.findFieldType(w.field)
	val := normalizeCondValue(w.value, ft)
	switch w.op {
	case "=", "==", "eq":
		return core.Eq(w.field, val, types.FieldType(0))
	case "!=", "<>":
		return core.Not(core.Eq(w.field, val, types.FieldType(0)))
	case ">", "gt":
		return core.Gt(w.field, val, types.FieldType(0))
	case ">=", "ge", "gte":
		return core.Ge(w.field, val, types.FieldType(0))
	case "<", "lt":
		return core.Lt(w.field, val, types.FieldType(0))
	case "<=", "le", "lte":
		return core.Le(w.field, val, types.FieldType(0))
	case "like":
		if s, ok := w.value.(string); ok {
			return core.Like(w.field, s)
		}
		return core.Eq(w.field, val, types.FieldType(0))
	case "in", "In", "IN":
		// 值须是切片/数组; 每个分量先经 normalizeCondValue 归一成列期望的
		// Go 类型再交给 core.In (分量类型不一会按列类型逐个编码比较)。
		var vals []interface{}
		rv := reflect.ValueOf(w.value)
		switch rv.Kind() {
		case reflect.Slice, reflect.Array:
			vals = make([]interface{}, rv.Len())
			for i := 0; i < rv.Len(); i++ {
				vals[i] = normalizeCondValue(rv.Index(i).Interface(), ft)
			}
		default:
			// 单值容忍: Where("Age","in",25) ≡ Where("Age","=",25)
			return core.Eq(w.field, val, types.FieldType(0))
		}
		if len(vals) == 0 {
			// 空集合恒假: InCondition.Match 对每行返回 false (无索引特化路径,
			// 不会被 nil 条件的"跳过过滤"语义坑成查全表)。
			return core.In(w.field, nil, types.FieldType(0))
		}
		return core.In(w.field, vals, types.FieldType(0))
	default:
		return core.Eq(w.field, val, types.FieldType(0))
	}
}

// normalizeCondValue 将条件值转换为 core 期望的类型
func normalizeCondValue(v interface{}, ft types.FieldType) interface{} {
	switch ft {
	case types.FieldType_Uint64:
		return toUint64(v)
	case types.FieldType_Uint32:
		return toInt32(v)
	case types.FieldType_Uint16:
		return int16(toUint64(v))
	case types.FieldType_Uint8:
		return uint8(toUint64(v))
	case types.FieldType_Int8:
		return toInt8(v)
	case types.FieldType_Int16:
		return toInt16(v)
	case types.FieldType_Int32:
		return toInt32(v)
	case types.FieldType_Int64:
		return toInt64(v)
	case types.FieldType_Float64:
		// 不能走 toUint64: 那会把条件值的小数先砍掉, 变成在浮点列上按整数匹配
		return toFloatVal(v)
	// ===== 扩展类型 =====
	//
	// 全部原样下传, 交给 core 的 types.EncodeValue / DecimalFromValue 解析与校验。
	// 尤其是 Decimal / UUID: 条件值常常是字符串 ("1.50" / "550e8400-..."), 一旦落进
	// 下面的 default 分支会被当定长串补零成定长字节, 比较和索引键就全错了。
	case types.FieldType_Bool,
		types.FieldType_Float32,
		types.FieldType_Decimal,
		types.FieldType_UUID,
		types.FieldType_Year,
		types.FieldType_Int128,
		types.FieldType_Blob:
		return v
	default:
		// 定长字符串类型: 原样下传, 由 core 按【列的实际宽度】编码
		// (types.EncodeValue 接受 string, 并顺带成为宽度闸门: 超宽报错而不是截断)。
		//
		// 不能在这里按 types.GetSize(ft) 自己补零: 那只是类型默认宽度, 对用宽度码声明的
		// 列 (maxlen:1000 → 1024 字节档) 会编出 256 字节——与行内字节不同源, 比较与索引键
		// 全错。而且这里只有 ft 没有列级参数, 本就拼不出正确的宽度。
		return v
	}
}

// findFieldType 查找字段的 FieldType
// 优先从 DB schema 查找，回退到 struct model
func (q *Query[T]) findFieldType(name string) types.FieldType {
	// 优先从 DB schema 查找
	if ft, ok := q.table.coreTable.GetActiveFieldType(name); ok {
		return ft
	}
	// 回退到 struct model
	for _, fd := range q.table.model.AllFields {
		if fd.ColName == name || fd.Name == name {
			return fd.FieldType
		}
	}
	return types.FieldType_Uint64 // 默认
}