package migrate

// ============================ 文本列采样定宽 / 字符列超档升档 ============================
//
// 【问题 1】inferSQLType 把 TEXT 家族 (TEXT/TINYTEXT/MEDIUMTEXT/LONGTEXT/CLOB/NCLOB)
// 一律落成 FieldType_Dynamicbyte: 动态列不能建索引 (引擎直接拒, 见 vrtx.go 建索引
// 那段的 ErrIndexDynamicField 处理), 读要走堆 —— 而真实库里很多 TEXT 列装的就是
// 几十几百字节的短文本 (备注 / 状态 / 标签), 白白失去索引能力。
//
// 【问题 2】带声明长度的 CHAR/VARCHAR(n) 按 bucketColumn 取定长档, 但 SQLite 不强制
// 声明长度 (MySQL 强制, 数据不可能超档) —— DockerManager 实测: sqlite 的 sys_ssl.cert
// 声明 VARCHAR(2048), 实际值 4803 字节, 按声明建 2048 定长档后写入即报"值 4803 字节
// 超过定长列容量 2048"。
//
// 【处理】Options.TextWidthScan (默认开) 让迁移在建目标表之前先【采样】: 对源表跑一次
// SELECT MAX(LENGTH(col)) (两路候选 —— TEXT 家族动态列 + 声明长度字符列 —— 同表合
// 一句, 每表一次), 按采样到的实际最大长度三态分流:
//   - 不超过 types.MaxInlineColumnWidth (2048) 的向上取 2 的幂 (下限 textWidthMin)
//     落成定长 VARCHAR 档 (可索引、读不走堆);
//   - 超过 2048 但不超过单值上限 types.MaxStringSize (16360) 的保持/升为动态列;
//   - 超过单值上限的落 Blob 链式列 (行内 8B 引用 + 链页, 上限约 4GB) —— 动态列按
//     单块上限拒写, 保持动态列会在【写入时】整表失败, 落 Blob 才能迁得动。
//
// 声明长度字符列 (declaredCharWidth 那一路) 多一层"当前档宽"对照: 采样值没超当前
// 档宽就不动 (声明兑现), 超了才按上面三态升档 (见 scanCharWidthUpgrade)。
//
// 【权衡】采样反映的是【当下】的数据: 定宽列会约束未来写入 (超宽即报错), 2 次幂圆整
// 最多浪费一倍行内空间; 且采样按表做一次, 行数大时是一次额外的全表扫描。在意这些就
// 关掉开关, 退回"TEXT 全落动态列 / 字符列按声明取档"的旧行为。
//
// 【为什么判定看源声明 (Column.Decl) 而不是看目标列类型】目标是动态列的不止 TEXT:
// JSON / TIME / 未知的兜底类型也都落动态列, 而它们不该被采样改宽 —— 采样只对"源库
// 明确说了这是文本"的列生效: TEXT 家族看 isTextFamilyDecl, 声明长度字符列看
// declaredCharWidth, 两者都从声明原文判定。
//
// 【为什么用可选接口而不是给 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), 再往上是动态列 / Blob 链式列的两档分流
// (见 textWidthCode)。
const textWidthMin = 32

// isTextFamilyDecl 源声明是否 TEXT 家族 (TEXT/TINYTEXT/MEDIUMTEXT/LONGTEXT/CLOB/NCLOB)。
// 用 splitTypeArgs 取基类型名: JSON / BLOB / BINARY / VARBINARY 都不在里面 —— 它们
// 各有各的归宿 (JSON 保持动态列, 二进制落 Blob 类型), 不参与采样。带长度的
// "TEXT(65535)" 声明 (MySQL 的 information_schema 会给 TEXT 报一个类型自身上界)
// 同样算 TEXT 家族 —— 那是类型的上界, 不是用户声明的宽度。带长度的 CHAR/VARCHAR
// 也不在这里: 那一路由 declaredCharWidth 单独判定 (取档规则不同, 见
// scanCharWidthUpgrade)。
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
}

// declaredCharWidth 源声明为【带长度的字符列】时返回声明字节数 args[0], 其余声明
// (无长度 / 非 CHAR/VARCHAR 家族) 返回 0。基类型名单与 inferSQLType 的字符列分支
// 同源: 这族声明按 bucketColumn 取定长档, 但 SQLite 不强制声明长度, 实际数据可能
// 超档 (DockerManager 实测 sys_ssl.cert VARCHAR(2048) 存 4803 字节) —— 声明字节数
// 就是"承诺兑现"的对照线, 采样超了它升档 (见 scanCharWidthUpgrade)。MySQL 侧读的
// 是 CHARACTER_OCTET_LENGTH (字节数), 同一函数兼容两种来源。
func declaredCharWidth(decl string) int {
	base, args := splitTypeArgs(strings.ToUpper(strings.TrimSpace(decl)))
	switch base {
	case "CHAR", "CHARACTER", "VARCHAR", "NCHAR", "NVARCHAR", "VARCHAR2":
	default:
		return 0
	}
	if len(args) == 0 {
		return 0
	}
	return args[0]
}

// textWidthClass 采样最大长度的三态归宿。
type textWidthClass int

const (
	// widthToFixed: ≤ MaxInlineColumnWidth, 落成定长 VARCHAR 档。
	widthToFixed textWidthClass = iota
	// widthToDynamic: 定长档装不下但还在动态列单值上限内, 保持动态列。
	widthToDynamic
	// widthToBlob: 超过动态列单值上限 (types.MaxStringSize), 落 Blob 链式列。
	widthToBlob
)

// textWidthCode 采样最大长度 → (宽度码, 档位宽度, 三态归宿):
//
//	maxLen ≤ MaxInlineColumnWidth              → 定长档: 大于等于 maxLen 的最小 2 次幂
//	             (下限 textWidthMin) 的宽度码;
//	MaxInlineColumnWidth < maxLen ≤ MaxStringSize → 动态列 (单值仍装得下, 只是不可定宽);
//	maxLen > MaxStringSize                     → Blob 链式列 (动态列按单块上限拒写,
//	                                             保持动态列会在写入时失败)。
//
// 2 的幂档 (32/64/.../2048) 全部精确落在引擎的 stringWidthTable 上, 所以
// types.WidthCodeFor 正好命中 (不会向上漂到 48/96 这种中间档): "采样 100 字节 →
// 128 档"是精确陈述, 不是约数。
func textWidthCode(maxLen int64) (code byte, width int, class textWidthClass) {
	switch {
	case maxLen > types.MaxStringSize:
		return 0, 0, widthToBlob
	case maxLen > types.MaxInlineColumnWidth:
		return 0, 0, widthToDynamic
	}
	w := textWidthMin
	for int64(w) < maxLen {
		w *= 2
	}
	return types.WidthCodeFor(w), w, widthToFixed
}

// scanTextWidths 宽度采样主逻辑, 建目标表之前调用。两路候选 (见
// isTextWidthCandidate) 合进同一次 per-table 采样:
//
//   - TEXT 家族动态列: 按三态归宿改类型 (见 textWidthCode) —— 够窄的改落
//     FieldType_String + 宽度码 (可索引), 中间的保持动态列, 超过动态列单值上限的落
//     FieldType_Blob (链式存储, 否则写入时必败);
//   - 声明长度字符列: 实际数据超当前档宽的按同一套三态升档 (见 scanCharWidthUpgrade),
//     没超档宽不动 (声明兑现)。
//
// 表结构就地修改, 返回写进迁移报告的说明 (无命中列时返回 "")。
//
// 放在 prepareForVRTX / fitRowWidthForVRTX 【之前】: 原主键列退化成普通列时要不要
// 补索引 (ensureColumnIndex 只给非动态列补), 以及行宽闸门要不要把采样得过宽的列
// 再降回动态列, 都该看到采样后的最终列宽。
//
// 采样失败不判死: 记日志并退回旧行为 (保持推断类型), 由返回值在报告里说明 —— 采样
// 是"让列更好用"的优化, 不该把以前能迁的表拦下来。
func scanTextWidths(src Source, sc *TableSchema, o *Options) string {
	var cols []string
	for i := range sc.Columns {
		if isTextWidthCandidate(&sc.Columns[i]) {
			cols = append(cols, sc.Columns[i].Name)
		}
	}
	// 不能提前 return: NUMERIC 域探测 (scanNumericDomain) 在采样列缺席时仍要跑。
	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 的宽度采样失败, 保持推断类型 (旧行为): %v", sc.Name, err)
		return appendNote(fmt.Sprintf("text 采样失败保持推断类型 (%v)", err), scanNumericDomain(src, sc, o))
	}
	var fixed, kept, blobbed, upgraded []string
	for i := range sc.Columns {
		c := &sc.Columns[i]
		maxLen := maxLens[c.Name] // 缺项视为 0, 全 NULL 的列落最窄档
		if c.Type != types.FieldType_Dynamicbyte || !isTextFamilyDecl(c.Decl) {
			// TEXT 家族以外走字符列那一路 (非候选列在里面自行排除)。
			if up := scanCharWidthUpgrade(c, maxLen, o); up != "" {
				upgraded = append(upgraded, up)
			}
			continue
		}
		code, width, class := textWidthCode(maxLen)
		switch class {
		case widthToBlob:
			// 采样值超过动态列单值上限: 保持动态列只会在【写入时】被引擎按单块上限
			// 拒掉 (整表迁移失败, 见 DockerManager 实测的 sys_logs.param 59566 字节),
			// 落 Blob 链式列才有出路: 行内 8B BlobRef, 内容可跨页, 单值约 4GB。
			// 代价照实报: 该列失去索引能力 (源列在 TEXT 上的索引会被 CreateTable 跳过)。
			c.Type = types.FieldType_Blob
			c.Param = 0
			blobbed = append(blobbed, fmt.Sprintf("%s (%d 字节)", c.Name, maxLen))
			o.vlogf("[migrate] 表 %s 的 TEXT 列 %s 采样最大 %d 字节, 超动态列单值上限 %d, 落 Blob 链式列",
				sc.Name, c.Name, maxLen, types.MaxStringSize)
		case widthToDynamic:
			kept = append(kept, fmt.Sprintf("%s 超 %d 字节", c.Name, types.MaxInlineColumnWidth))
		default:
			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, ", ")+")")
	}
	if len(blobbed) > 0 {
		note = appendNote(note, fmt.Sprintf("text 列超 %d 字节落 Blob 链式列 (%s)",
			types.MaxStringSize, strings.Join(blobbed, ", ")))
	}
	if len(upgraded) > 0 {
		note = appendNote(note, "声明长度字符列超档升档 ("+strings.Join(upgraded, "; ")+")")
	}
	return appendNote(note, scanNumericDomain(src, sc, o))
}

// isTextWidthCandidate 这一列是否参与建表前的宽度采样。两路与源声明一一对应:
//   - TEXT 家族动态列 (isTextFamilyDecl): 采样实际最大长度, 够窄的从动态列【升】定长档;
//   - 声明长度字符列 (declaredCharWidth): 定长档装不下实际数据时【升档】兜住。
//
// 其余列各有各的归宿, 不碰: JSON / TIME / 未知兜底类型保持动态列, 二进制落 Blob 类型
// (声明多大的容量都兑现得了, v10 起 Blob 行内是 8B 链引用、单值约 4GB), 无长度字符
// 声明按 32 字节档处理 —— 它们都没有"声明了宽度却装不下"的落差。
func isTextWidthCandidate(c *Column) bool {
	if c.Type == types.FieldType_Dynamicbyte && isTextFamilyDecl(c.Decl) {
		return true
	}
	// 声明长度字符列: 定长档 (FieldType_String) 的核对档宽; 声明超 2048 落成动态列
	// 的也要 —— 那种列只剩"超单值上限落 Blob"一种升档, 不采样就会在写入时整表失败
	// (与 TEXT 家族超单值上限落 Blob 是同一个坑)。
	return declaredCharWidth(c.Decl) > 0 &&
		(c.Type == types.FieldType_String || c.Type == types.FieldType_Dynamicbyte)
}

// scanCharWidthUpgrade 声明长度字符列的【超档升档】: 采样 maxLen 没超当前档宽就不动
// (声明兑现; MySQL 强制声明长度, 数据不可能超档, 恒走这条, 所以这路对 MySQL 无害);
// 超了按 textWidthCode 三态升档 —— 能定长则升 2 次幂定长档 (上限 2048), 超 2048 落
// 动态列, 超 types.MaxStringSize (16360) 落 Blob 链式列 (动态列按单块上限拒写, 不升
// Blob 这类大值会在写入时让整表迁移失败)。就地修改列类型, 返回写进迁移报告的说明
// (无升档时返回 "")。
func scanCharWidthUpgrade(c *Column, maxLen int64, o *Options) string {
	declared := declaredCharWidth(c.Decl)
	if declared <= 0 ||
		(c.Type != types.FieldType_String && c.Type != types.FieldType_Dynamicbyte) {
		return ""
	}
	// 当前档宽: 定长列看宽度码对应的档位; 已是动态列的按单值上限算 (动态列的"档"
	// 就是整个堆页, 装得下任何不超单值上限的值) —— 那种列只剩超单值上限落 Blob
	// 这一种升档, textWidthCode 自然会给出。
	curWidth := types.MaxStringSize
	if c.Type == types.FieldType_String {
		curWidth = int(types.FieldSize(c.Type, c.Param))
	}
	if maxLen <= int64(curWidth) {
		return "" // 声明兑现: 实际数据装得进当前档, 不动。
	}
	code, width, class := textWidthCode(maxLen)
	var target string
	switch class {
	case widthToBlob:
		c.Type = types.FieldType_Blob
		c.Param = 0
		target = "Blob 链式列"
	case widthToDynamic:
		c.Type = types.FieldType_Dynamicbyte
		c.Param = 0
		target = "动态列"
	default:
		c.Type = types.FieldType_String
		c.Param = code
		target = fmt.Sprintf("%d 字节档", width)
	}
	o.vlogf("[migrate] 字符列 %s 声明 %d 字节但采样最大 %d 字节, 升档 %s",
		c.Name, declared, maxLen, target)
	return fmt.Sprintf("列 %s 声明 %d 字节但实际最大 %d 字节, 升档 %s", c.Name, declared, maxLen, target)
}

// 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, ", ") + ")"
}