package migrate

// ============================ TEXT 列采样定宽 ============================
//
// 【问题】inferSQLType 把 TEXT 家族 (TEXT/TINYTEXT/MEDIUMTEXT/LONGTEXT/CLOB/NCLOB)
// 一律落成 FieldType_Dynamicbyte: 动态列不能建索引 (引擎直接拒, 见 vrtx.go 建索引
// 那段的 ErrIndexDynamicField 处理), 读要走堆 —— 而真实库里很多 TEXT 列装的就是
// 几十几百字节的短文本 (备注 / 状态 / 标签), 白白失去索引能力。
//
// 【处理】Options.TextWidthScan (默认开) 让迁移在建目标表之前先【采样】: 对源表跑一次
// SELECT MAX(LENGTH(col)) (同表多列合一句, 每表一次), 按采样到的实际最大长度向上取
// 2 的幂 (下限 textWidthMin, 上限 types.MaxInlineColumnWidth) 落成定长 VARCHAR 档;
// 采样超过 2048 字节的列保持动态列。定宽列可建索引、读不走堆。
//
// 【权衡】采样反映的是【当下】的数据: 定宽列会约束未来写入 (超宽即报错), 2 次幂圆整
// 最多浪费一倍行内空间; 且采样按表做一次, 行数大时是一次额外的全表扫描。在意这些就
// 关掉开关, 退回"TEXT 全落动态列"的旧行为。
//
// 【为什么判定看源声明 (Column.Decl) 而不是看目标列类型】目标是动态列的不止 TEXT:
// JSON / TIME / 未知的兜底类型也都落动态列, 而它们不该被采样改宽 —— 采样定宽只对
// "源库明确说了这是无长度文本"的列生效。带长度的 CHAR/VARCHAR 按声明容量取档
// (bucketColumn), 也不经过这里。
//
// 【为什么用可选接口而不是给 Source 加方法】采样需要 SQL 查询通道, 只有 *SQL 有;
// VRTX 源 (VRTX → VRTX 拷贝) 没有这一层, 也不会声明 TEXT 列。给 Source 接口加方法
// 会强迫 VRTX 实现一个它根本用不上的能力, 断言式可选接口 (TextWidthSampler) 让
// 不支持采样的源端自然退回旧行为。

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

// TextWidthSampler 源端可选能力: 建目标表前的数据采样 (只有 *SQL 有 SQL 查询通道;
// VRTX 源不实现, 自然退回旧行为)。
type TextWidthSampler interface {
	// TextWidths 返回每列的 MAX(LENGTH(col)), 全 NULL / 空表为 0。
	// 同表多列应一次查询取回。
	TextWidths(table string, cols []string) (map[string]int64, error)
	// NumericDomain 返回每列【域外值】的个数: 非 NULL 且取值不在 {0, 1} 的行数。
	// 用于裸 NUMERIC → Bool 的数据兜底 (见 scanTextWidths 的 NUMERIC 探测);
	// 同表多列应一次查询取回。全 NULL / 全 0/1 / 空表返回 0。
	NumericDomain(table string, cols []string) (map[string]int64, error)
}

// textWidthMin 采样定宽的档位下限 (字节)。TEXT 列哪怕当前全空也落 32 档而不是 16 —
// 16 字节装不下一个像样的人名/备注, 而 32 只多付 16 字节; 上限就是行内宽度上限
// types.MaxInlineColumnWidth (2048): 采样超过它只能保持动态列。
const textWidthMin = 32

// isTextFamilyDecl 源声明是否 TEXT 家族 (TEXT/TINYTEXT/MEDIUMTEXT/LONGTEXT/CLOB/NCLOB)。
// 用 splitTypeArgs 取基类型名: JSON / BLOB / BINARY / VARBINARY 与带长度的 CHAR /
// VARCHAR 都不在里面 —— 它们各有各的归宿 (JSON 保持动态列, 二进制落 Blob 类型,
// 声明长度的字符列按声明容量取档), 不参与采样。带长度的 "TEXT(65535)" 声明
// (MySQL 的 information_schema 会给 TEXT 报一个类型自身上界) 同样算 TEXT 家族 —
// 那是类型的上界, 不是用户声明的宽度。
func isTextFamilyDecl(decl string) bool {
	base, _ := splitTypeArgs(strings.ToUpper(strings.TrimSpace(decl)))
	switch base {
	case "TEXT", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT", "CLOB", "NCLOB":
		return true
	}
	return false
}

// textWidthCode 采样最大长度 → 定长档位: 大于等于 maxLen 的最小 2 次幂 (下限
// textWidthMin, 上限 types.MaxInlineColumnWidth), 返回 (宽度码, 档位宽度, 可否定宽);
// maxLen 超上限返回 ok=false —— 那一列只能保持动态列。
//
// 2 的幂档 (32/64/.../2048) 全部精确落在引擎的 stringWidthTable 上, 所以
// types.WidthCodeFor 正好命中 (不会向上漂到 48/96 这种中间档): "采样 100 字节 →
// 128 档"是精确陈述, 不是约数。
func textWidthCode(maxLen int64) (code byte, width int, ok bool) {
	if maxLen > types.MaxInlineColumnWidth {
		return 0, 0, false
	}
	w := textWidthMin
	for int64(w) < maxLen {
		w *= 2
	}
	return types.WidthCodeFor(w), w, true
}

// scanTextWidths TEXT 采样定宽主逻辑, 建目标表之前调用: 找出 schema 里"源声明
// TEXT 家族、当前推断为动态列"的列, 让源端按表采样一次, 够窄的改落成
// FieldType_String + 宽度码 (可索引), 超宽的保持动态列。表结构就地修改, 返回写进
// 迁移报告的说明 (无命中列时返回 "")。
//
// 放在 prepareForVRTX / fitRowWidthForVRTX 【之前】: 原主键列退化成普通列时要不要
// 补索引 (ensureColumnIndex 只给非动态列补), 以及行宽闸门要不要把采样得过宽的列
// 再降回动态列, 都该看到采样后的最终列宽。
//
// 采样失败不判死: 记日志并退回旧行为 (全落动态列), 由返回值在报告里说明 —— 采样
// 是"让列更好用"的优化, 不该把以前能迁的表拦下来。
func scanTextWidths(src Source, sc *TableSchema, o *Options) string {
	var cols []string
	for i := range sc.Columns {
		c := &sc.Columns[i]
		if c.Type != types.FieldType_Dynamicbyte || !isTextFamilyDecl(c.Decl) {
			continue
		}
		cols = append(cols, c.Name)
	}
	// 不能提前 return: NUMERIC 域探测 (scanNumericDomain) 在 TEXT 列缺席时仍要跑。
	if len(cols) == 0 {
		return scanNumericDomain(src, sc, o)
	}
	sampler, ok := src.(TextWidthSampler)
	if !ok {
		return scanNumericDomain(src, sc, o) // 源端不支持采样 (VRTX → VRTX): 两条路都退
	}
	maxLens, err := sampler.TextWidths(sc.Name, cols)
	if err != nil {
		o.logf("[migrate] 表 %s 的 TEXT 采样失败, 保持动态列 (旧行为): %v", sc.Name, err)
		return appendNote(fmt.Sprintf("text 采样失败保持动态列 (%v)", err), scanNumericDomain(src, sc, o))
	}
	var fixed, kept []string
	for i := range sc.Columns {
		c := &sc.Columns[i]
		if c.Type != types.FieldType_Dynamicbyte || !isTextFamilyDecl(c.Decl) {
			continue
		}
		maxLen := maxLens[c.Name] // 缺项视为 0, 全 NULL 的列落最窄档
		code, width, ok := textWidthCode(maxLen)
		if !ok {
			kept = append(kept, fmt.Sprintf("%s 超 %d 字节", c.Name, types.MaxInlineColumnWidth))
			continue
		}
		c.Type = types.FieldType_String
		c.Param = code
		fixed = append(fixed, fmt.Sprintf("%s %d 字节", c.Name, width))
		o.vlogf("[migrate] 表 %s 的 TEXT 列 %s 采样最大 %d 字节, 定宽 %d 字节",
			sc.Name, c.Name, maxLen, width)
	}
	var note string
	if len(fixed) > 0 {
		note = appendNote(note, "text 列按采样定宽 ("+strings.Join(fixed, ", ")+")")
	}
	if len(kept) > 0 {
		note = appendNote(note, "text 列采样超宽保持动态 ("+strings.Join(kept, ", ")+")")
	}
	return appendNote(note, scanNumericDomain(src, sc, o))
}

// isBareNumericDecl 源声明是否【裸 NUMERIC】(基类型 NUMERIC 且无 (p,s) 参数)。
// 只有这种声明会被 inferSQLType (SQLite 分支) 收成 Bool; 它同时是数据探测的
// 目标 — 真定点列要么带精度 NUMERIC(p,s), 要么写 DECIMAL, 都不进这里。
func isBareNumericDecl(decl string) bool {
	base, args := splitTypeArgs(strings.ToUpper(strings.TrimSpace(decl)))
	return base == "NUMERIC" && len(args) == 0
}

// scanNumericDomain 裸 NUMERIC → Bool 的数据兜底: inferSQLType 按"GORM sqlite
// 方言的 bool 声明是裸 NUMERIC"这一高概率先收成 Bool, 但 SQLite 的 NUMERIC 是
// 亲和性而非真类型 — 手写的定点列也可能裸声明。建表前采样一次域外值: 存在
// 非 {0,1,NULL} 的行就把该列改回 Decimal(0) (裸 NUMERIC 的旧映射), 并写进
// 报告; 探测失败保持 Bool 不动 (bool 仍是高概率, 且写入侧对 Bool 列写非 0/1
// 只会拒值, 不会静默错)。返回报告 note, 无命中列时返回 ""。
func scanNumericDomain(src Source, sc *TableSchema, o *Options) string {
	var cols []string
	for i := range sc.Columns {
		c := &sc.Columns[i]
		if c.Type == types.FieldType_Bool && isBareNumericDecl(c.Decl) {
			cols = append(cols, c.Name)
		}
	}
	if len(cols) == 0 {
		return ""
	}
	sampler, ok := src.(TextWidthSampler)
	if !ok {
		return ""
	}
	od, err := sampler.NumericDomain(sc.Name, cols)
	if err != nil {
		o.logf("[migrate] 表 %s 的 NUMERIC 域探测失败, 裸 NUMERIC 列保持 Bool: %v", sc.Name, err)
		return fmt.Sprintf("numeric 域探测失败保持 Bool (%v)", err)
	}
	var reverted []string
	for i := range sc.Columns {
		c := &sc.Columns[i]
		if c.Type == types.FieldType_Bool && isBareNumericDecl(c.Decl) && od[c.Name] > 0 {
			// 域外值说明这列真不是布尔。旧映射 Decimal(0) 对带小数的域外值
			// (1.5 等) 照样拒写 — 兜底不该把能迁的表拦下, 退 float64 (有损,
			// 与 DECIMAL 超范围时的既有降级同 philosophy, 报告 note 说明)。
			c.Type = types.FieldType_Float64
			c.Param = 0
			reverted = append(reverted, fmt.Sprintf("%s (%d 行域外值, 降级 float64)", c.Name, od[c.Name]))
			o.vlogf("[migrate] 表 %s 的裸 NUMERIC 列 %s 含 %d 行非布尔值, 降级 float64",
				sc.Name, c.Name, od[c.Name])
		}
	}
	if len(reverted) == 0 {
		return ""
	}
	return "裸 NUMERIC 探测到域外值降级 float64 (" + strings.Join(reverted, ", ") + ")"
}