what
21b80bdea4
- exec/scanner: 用 *interface{} 替换 **json.RawMessage 扫描目标，兼容 DuckDB 返回 map[string]interface{} 的场景；新增 toJSONRawMessage 转换函数
- exec/scanner: ScanVal 支持结构体指针，通过 JSON 中间层转换（DuckDB STRUCT 列）
- exec/scanner: 将 *sql.RawBytes 和 *[]byte 的处理从 ScanValContext 移入 scanner.ScanVal
- exec/query_executor: 简化 ScanValContext，移除私有 scan 方法
- exec: 补充 scanner 级别 ScanVal 测试用例
- internal/util/reflect: 重写 SafeSetVarValue，修复非指针 src 及 nil 指针字段的 panic
- internal/util/column_map: 恢复非匿名带标签结构体字段的展开逻辑
- schema: 新增 vector 列类型支持
- engine: 补充 DuckDB 相关配置
- dialect/sqlite3/vtab: 完善虚拟表适配器
- 各方言测试改用 sqlmock 虚拟连接
263 lines
9.2 KiB
Go
263 lines
9.2 KiB
Go
//go:build sqlite_vtable || vtable
|
||
|
||
package vtab
|
||
|
||
import (
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
gosqlite3 "github.com/mattn/go-sqlite3"
|
||
)
|
||
|
||
// ─── 桥接层：Module → gosqlite3.Module ──────────────────────
|
||
|
||
type moduleAdapter struct {
|
||
mod Module
|
||
}
|
||
|
||
func (a *moduleAdapter) Create(c *gosqlite3.SQLiteConn, args []string) (gosqlite3.VTab, error) {
|
||
return a.build(c, args, true)
|
||
}
|
||
|
||
func (a *moduleAdapter) Connect(c *gosqlite3.SQLiteConn, args []string) (gosqlite3.VTab, error) {
|
||
return a.build(c, args, false)
|
||
}
|
||
|
||
func (a *moduleAdapter) DestroyModule() {}
|
||
|
||
func (a *moduleAdapter) build(c *gosqlite3.SQLiteConn, args []string, isCreate bool) (gosqlite3.VTab, error) {
|
||
declare := func(schema string) error { return c.DeclareVTab(schema) }
|
||
|
||
var (
|
||
table Table
|
||
err error
|
||
)
|
||
if isCreate {
|
||
table, err = a.mod.Create(args, declare)
|
||
} else {
|
||
table, err = a.mod.Connect(args, declare)
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
base := &baseVtabAdapter{table: table}
|
||
if wt, ok := table.(WritableTable); ok {
|
||
return &writableVtabAdapter{baseVtabAdapter: base, wt: wt}, nil
|
||
}
|
||
return base, nil
|
||
}
|
||
|
||
func NewModuleAdapter(mod Module) gosqlite3.Module {
|
||
return &moduleAdapter{mod}
|
||
}
|
||
|
||
// ─── 桥接层：Table（只读）────────────────────────────────────
|
||
|
||
// planKey 是查询计划的唯一标识，由 BestIndex 写入、Filter 读取。
|
||
// 两个字段均由用户实现的 BestIndex 返回，SQLite 原样透传给 Filter，
|
||
// 组合唯一对应一份约束元数据列表（列索引 + 操作符）。
|
||
type planKey struct {
|
||
idxNum int // 对应 IndexOutput.IdxNum，用于区分不同查询计划
|
||
idxStr string // 对应 IndexOutput.IdxStr，与 idxNum 配合进一步区分计划。
|
||
// 与 SQL 字段名无关，是 BestIndex → Filter 之间的自由通信通道，常见用途：
|
||
// 1. 传递索引名（如 "idx_name"），告知 Filter 按哪个索引逻辑过滤；
|
||
// 2. 序列化约束条件，Filter 直接解析，省去查 plans map 的步骤；
|
||
// 3. 传递排序方向（如 "asc"/"desc"）。
|
||
// SQLite 不解释其含义，仅原样透传给 Filter。当前实现固定返回 ""，未使用。
|
||
}
|
||
|
||
type baseVtabAdapter struct {
|
||
table Table
|
||
// plans 保存每个查询计划中 Used=true 的约束（按值传入顺序），
|
||
// BestIndex 写入，Filter 通过 (idxNum, idxStr) 查询后绑定值。
|
||
plans map[planKey][]ConstraintInfo
|
||
}
|
||
|
||
// BestIndex 是适配层的 xBestIndex 实现，将 go-sqlite3 的 C 结构转换为 Go 接口，
|
||
// 调用用户实现的 table.BestIndex，再把结果保存为"查询计划"供 Filter 阶段使用。
|
||
//
|
||
// 调用时机：每次 SQLite 准备执行针对此虚拟表的查询时调用，可能调用多次（不同约束组合）。
|
||
// 执行顺序：BestIndex → （SQLite 选定计划）→ Filter（实际执行查询）。
|
||
//
|
||
// SQLite 对同一个 prepared statement 可以不重新调用 BestIndex
|
||
func (v *baseVtabAdapter) BestIndex(csts []gosqlite3.InfoConstraint, obs []gosqlite3.InfoOrderBy) (*gosqlite3.IndexResult, error) {
|
||
// 将 go-sqlite3 的 C 结构转换为 Go 的 ConstraintInfo，传给用户实现。
|
||
// 每个 constraint 对应 WHERE 子句中的一个条件（列、操作符、是否可用）。
|
||
ci := make([]ConstraintInfo, len(csts))
|
||
for i, c := range csts {
|
||
ci[i] = ConstraintInfo{Column: c.Column, Op: c.Op, Usable: c.Usable}
|
||
}
|
||
ob := make([]OrderByInfo, len(obs))
|
||
for i, o := range obs {
|
||
ob[i] = OrderByInfo{Column: o.Column, Desc: o.Desc}
|
||
}
|
||
|
||
out, err := v.table.BestIndex(ci, ob)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// adapter 统一接管所有 Usable 的约束，用户无需在 IndexOutput 声明 Used。
|
||
// Filter 阶段 SQLite 只传入 Used=true 的约束值（argv），
|
||
// 需要靠这里保存的顺序和列信息才能还原出完整的 ConstraintInfo。
|
||
used := make([]bool, len(ci))
|
||
var usedCi []ConstraintInfo
|
||
for i, c := range ci {
|
||
if c.Usable {
|
||
used[i] = true
|
||
usedCi = append(usedCi, c)
|
||
}
|
||
}
|
||
|
||
// 用约束的列索引+操作符自动生成唯一 planKey，确保不同查询类型的计划互不覆盖。
|
||
parts := make([]string, len(usedCi))
|
||
for i, c := range usedCi {
|
||
parts[i] = fmt.Sprintf("%d:%d", c.Column, int(c.Op))
|
||
}
|
||
idxStr := strings.Join(parts, ",")
|
||
idxNum := len(usedCi)
|
||
|
||
if v.plans == nil {
|
||
v.plans = make(map[planKey][]ConstraintInfo)
|
||
}
|
||
v.plans[planKey{idxNum, idxStr}] = usedCi
|
||
|
||
return &gosqlite3.IndexResult{
|
||
Used: used,
|
||
IdxNum: idxNum,
|
||
IdxStr: idxStr,
|
||
AlreadyOrdered: out.AlreadyOrdered,
|
||
EstimatedCost: out.EstimatedCost,
|
||
EstimatedRows: out.EstimatedRows,
|
||
}, nil
|
||
}
|
||
|
||
func (v *baseVtabAdapter) Disconnect() error { return v.table.Disconnect() }
|
||
func (v *baseVtabAdapter) Destroy() error { return v.table.Destroy() }
|
||
|
||
func (v *baseVtabAdapter) Open() (gosqlite3.VTabCursor, error) {
|
||
c, err := v.table.Open()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &cursorAdapter{cursor: c, plans: v.plans}, nil
|
||
}
|
||
|
||
// ─── 桥接层：WritableTable ────────────────────────────────────
|
||
|
||
// writableVtabAdapter 内嵌只读适配器并实现 gosqlite3.VTabUpdater，
|
||
// go-sqlite3 通过类型断言检测到该接口后会启用 INSERT/UPDATE/DELETE。
|
||
type writableVtabAdapter struct {
|
||
*baseVtabAdapter
|
||
wt WritableTable
|
||
}
|
||
|
||
func (v *writableVtabAdapter) Delete(rowid any) error {
|
||
return v.wt.Delete(rowid)
|
||
}
|
||
|
||
func (v *writableVtabAdapter) Insert(rowid any, values []any) (int64, error) {
|
||
// rowid 是 SQLite 建议的值（通常为 nil，由表自行决定）
|
||
return v.wt.Insert(values)
|
||
}
|
||
|
||
func (v *writableVtabAdapter) Update(rowid any, values []any) error {
|
||
return v.wt.Update(rowid, values)
|
||
}
|
||
|
||
// ─── 桥接层：Cursor ───────────────────────────────────────────
|
||
|
||
type cursorAdapter struct {
|
||
cursor Cursor
|
||
// plans 是 Open() 时从 adapter 复制的计划快照，与后续 BestIndex 调用隔离，
|
||
// 防止 list/item 等不同查询的 BestIndex 相互覆盖导致 Filter 拿到错误的列信息。
|
||
plans map[planKey][]ConstraintInfo
|
||
}
|
||
|
||
func (c *cursorAdapter) Close() error { return c.cursor.Close() }
|
||
func (c *cursorAdapter) Next() error { return c.cursor.Next() }
|
||
func (c *cursorAdapter) EOF() bool { return c.cursor.EOF() }
|
||
|
||
// Filter 是适配层的 xFilter 实现，将 SQLite 传入的约束值与 BestIndex 保存的计划合并，
|
||
// 还原出完整的 ConstraintInfo 列表后调用用户实现的 tCursor.Filter。
|
||
//
|
||
// 调用时机：每次实际执行查询时（包括 IN 展开的每一次子查询）。
|
||
// 执行顺序：BestIndex（保存计划）→ Filter（绑定值、执行查询）。
|
||
//
|
||
// Filter 的本质是把两份分离的信息合并成一条完整的过滤条件：
|
||
// ci[i].Column + ci[i].Op ←（BestIndex 存的：哪列、什么操作）
|
||
// +
|
||
// vals[i] ←（SQLite 传的：过滤值）
|
||
// =
|
||
// WHERE column OP value ←（最终传给 tCursor.Filter 的约束）
|
||
// 合并完之后交给 tCursor.Filter，再转成 fiter map，最终作为参数传给 vt.Select 去调 API。
|
||
func (c *cursorAdapter) Filter(idxNum int, idxStr string, vals []any) error {
|
||
// 通过 (idxNum, idxStr) 找到 BestIndex 阶段保存的约束元数据（列索引、操作符）。
|
||
// vals 只含约束的值，没有列和操作符信息，必须与 ci 对应位置合并才能还原完整约束。
|
||
ci := c.plans[planKey{idxNum, idxStr}]
|
||
|
||
// vals 长度 = BestIndex 中 Used=true 的约束数量（SQLite 保证一一对应）。
|
||
// ⚠️ SQLite 3.38+ IN 约束场景下，ci 可能比 vals 短（计划被 Usable=false 调用冲掉时）。
|
||
// 此时超出 ci 长度的约束 Column/Op 会是零值，由 tCursor.Filter 根据 Op 类型决定是否使用。
|
||
constraints := make([]ConstraintInfo, len(vals))
|
||
for i, val := range vals {
|
||
constraints[i].Value = val // 绑定 SQLite 传入的约束值
|
||
if i < len(ci) {
|
||
constraints[i].Column = ci[i].Column // 对应列索引，用于 GetColField
|
||
constraints[i].Op = ci[i].Op // 操作符，用于映射到 exp.BooleanOperation
|
||
constraints[i].Usable = true
|
||
}
|
||
// i >= len(ci)：计划缺失，Column=0/Op=0(OpIN)/Usable=false，
|
||
// tCursor.Filter 需针对 OpIN 单独放行（不依赖 Usable 判断）。
|
||
}
|
||
|
||
return c.cursor.Filter(idxNum, constraints)
|
||
}
|
||
|
||
func (c *cursorAdapter) Rowid() (int64, error) { return c.cursor.Rowid() }
|
||
|
||
func (c *cursorAdapter) Column(ctx *gosqlite3.SQLiteContext, col int) error {
|
||
val, err := c.cursor.Column(col)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
resultValue(ctx, val)
|
||
return nil
|
||
}
|
||
|
||
// resultValue 将 Go 值写入 SQLite 上下文。
|
||
func resultValue(ctx *gosqlite3.SQLiteContext, val any) {
|
||
if val == nil {
|
||
ctx.ResultNull()
|
||
return
|
||
}
|
||
switch v := val.(type) {
|
||
case int:
|
||
ctx.ResultInt(v)
|
||
case int32:
|
||
ctx.ResultInt(int(v))
|
||
case int64:
|
||
ctx.ResultInt64(v)
|
||
case float32:
|
||
ctx.ResultDouble(float64(v))
|
||
case float64:
|
||
ctx.ResultDouble(v)
|
||
case bool:
|
||
if v {
|
||
ctx.ResultInt(1)
|
||
} else {
|
||
ctx.ResultInt(0)
|
||
}
|
||
case time.Time:
|
||
ctx.ResultText(v.Format("2006-01-02 15:04:05"))
|
||
case string:
|
||
ctx.ResultText(v)
|
||
case []byte:
|
||
ctx.ResultBlob(v)
|
||
default:
|
||
ctx.ResultText(fmt.Sprint(v))
|
||
}
|
||
}
|