package enginetest

import (
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"testing"

	"github.com/dolthub/go-mysql-server/sql"
	"github.com/dolthub/vitess/go/vt/sqlparser"
	"github.com/lib/pq/oid"
	"github.com/stretchr/testify/require"

	"github.com/dolthub/doltgresql/postgres/parser/parser"
	"github.com/dolthub/doltgresql/postgres/parser/sem/tree"
	"github.com/dolthub/doltgresql/postgres/parser/types"
)

func convertQuery(query string) []string {
	if queries, converted := transformAST(query); converted {
		return queries
	}

	query = normalizeStrings(query)
	query = convertDoltProcedureCalls(query)
	return []string{query}
}

func transformAST(query string) ([]string, bool) {
	parser := sql.NewMysqlParser()
	stmt, err := parser.ParseSimple(query)
	if err != nil {
		return nil, false
	}

	switch stmt := stmt.(type) {
	case *sqlparser.DDL:
		switch stmt.Action {
		case "create":
			return transformCreateTable(stmt)
		case "drop":
			return transformDrop(query, stmt)
		case "rename":
			return transformRename(stmt)
		}
	case *sqlparser.Set:
		return transformSet(stmt)
	case *sqlparser.Select:
		return transformSelect(stmt)
	case *sqlparser.Insert:
		return transformInsert(stmt)
	case *sqlparser.Update:
		return transformUpdate(stmt)
	case *sqlparser.Delete:
		return transformDelete(stmt)
	case *sqlparser.AlterTable:
		return transformAlterTable(stmt)
	}

	return nil, false
}

// transformDelete rewrites DELETE statements that reference a `db.tbl`
// qualifier so the qualifier becomes postgres' `db.public.tbl`. We don't
// attempt any other DELETE rewriting today.
func transformDelete(stmt *sqlparser.Delete) ([]string, bool) {
	if containsQualifiedTableName(stmt) {
		return []string{formatNode(stmt)}, true
	}
	return nil, false
}

// transformUpdate handles MySQL UPDATE statements that need rewriting before
// Postgres can run them. Right now the only case we rewrite is `UPDATE IGNORE`:
// Postgres has no IGNORE keyword on UPDATE, but the closest approximation is
// `INSERT INTO t (cols) SELECT new_values FROM t WHERE cond ON CONFLICT DO NOTHING`,
// which silently skips rows that would violate a constraint. The semantics are
// not exactly equivalent (the original rows still exist, and the rowcount is
// reported as INSERT rows rather than UPDATE matched/changed), but it lets the
// tests at least exercise the conflict-handling code paths.
func transformUpdate(stmt *sqlparser.Update) ([]string, bool) {
	// Reformat plain UPDATEs that reference a `db.tbl` so the qualifier gets
	// rewritten to postgres' `db.public.tbl`. UPDATE IGNORE goes through the
	// INSERT … ON CONFLICT translation below regardless.
	if stmt.Ignore != "ignore " {
		if containsQualifiedTableName(stmt) {
			return []string{formatNode(stmt)}, true
		}
		return nil, false
	}

	// Only handle single-table UPDATE IGNORE — JOIN updates would need much more
	// translation work.
	if len(stmt.TableExprs) != 1 {
		return nil, false
	}
	aliased, ok := stmt.TableExprs[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return nil, false
	}
	tname, ok := aliased.Expr.(sqlparser.TableName)
	if !ok {
		return nil, false
	}
	tableName := translateTableName(tname)

	// Build the column list and the SELECT expressions from the SET clause.
	colList := make(tree.NameList, len(stmt.Exprs))
	selectExprs := make(tree.SelectExprs, len(stmt.Exprs))
	for i, e := range stmt.Exprs {
		colList[i] = tree.Name(e.Name.Name.String())
		selectExprs[i] = tree.SelectExpr{Expr: convertExpr(e.Expr)}
	}

	// FROM t [AS alias]
	fromTable := &tree.AliasedTableExpr{
		Expr: translateTableName(tname),
	}
	if !aliased.As.IsEmpty() {
		fromTable.As = tree.AliasClause{Alias: tree.Name(aliased.As.String())}
	}

	selectClause := &tree.SelectClause{
		Exprs: selectExprs,
		From:  tree.From{Tables: tree.TableExprs{fromTable}},
		Where: convertWhere(stmt.Where),
	}

	insert := tree.Insert{
		Table:   tableName,
		Columns: colList,
		Rows: &tree.Select{
			Select: selectClause,
		},
		OnConflict: &tree.OnConflict{
			Columns:   tree.NameList{tree.Name("fake")}, // placeholder, see transformInsert
			DoNothing: true,
		},
		Returning: &tree.NoReturningClause{},
	}

	ctx := formatNodeWithUnqualifiedTableNames(&insert)
	return []string{ctx.String()}, true
}

func transformRename(stmt *sqlparser.DDL) ([]string, bool) {
	rename := &tree.RenameTable{
		Name:    TableNameToUnresolvedObjectName(stmt.FromTables[0]),
		NewName: TableNameToUnresolvedObjectName(stmt.ToTables[0]),
	}

	ctx := formatNodeWithUnqualifiedTableNames(rename)
	return []string{ctx.String()}, true
}

func transformInsert(stmt *sqlparser.Insert) ([]string, bool) {
	table := stmt.Table
	// REPLACE INTO has no postgres equivalent. The closest match is
	// `INSERT ... ON CONFLICT (...) DO UPDATE SET col = <new value>`, which
	// updates an existing row in place (vs. DELETE + INSERT) but produces the
	// same end state. We use the literal values from the INSERT row as the
	// right-hand side (rather than EXCLUDED.col references, which doltgres
	// doesn't yet resolve back through its MySQL translation layer).
	//
	// Limitations:
	//   - REPLACE without an explicit column list — we'd need schema knowledge
	//     to enumerate the target columns. REPLACE ... SET form is parsed by
	//     vitess with Columns populated, so it works.
	//   - Multi-row REPLACE — a single SET clause can't refer to multiple
	//     candidate rows. Single-row only for now.
	if stmt.Action == sqlparser.ReplaceStr {
		if len(stmt.Columns) == 0 {
			return nil, false
		}

		// Pull a single row of values out of the parsed insert. Vitess wraps
		// `VALUES (...)` and `SET col = ...` in the same shape.
		var valTuple sqlparser.ValTuple
		switch r := stmt.Rows.(type) {
		case sqlparser.Values:
			if len(r) != 1 {
				return nil, false
			}
			valTuple = r[0]
		case *sqlparser.AliasedValues:
			if len(r.Values) != 1 {
				return nil, false
			}
			valTuple = r.Values[0]
		default:
			return nil, false
		}
		if len(valTuple) != len(stmt.Columns) {
			return nil, false
		}

		tableName := translateTableName(table)

		colList := make(tree.NameList, len(stmt.Columns))
		updateExprs := make(tree.UpdateExprs, len(stmt.Columns))
		for i, col := range stmt.Columns {
			name := tree.Name(col.String())
			colList[i] = name
			updateExprs[i] = &tree.UpdateExpr{
				Names: tree.NameList{name},
				Expr:  convertExpr(valTuple[i]),
			}
		}

		insert := tree.Insert{
			Table:   tableName,
			Columns: colList,
			Rows:    rowsForInsert(stmt.Rows),
			OnConflict: &tree.OnConflict{
				Columns: tree.NameList{tree.Name("fake")}, // placeholder; doltgres ignores the conflict-target list
				Exprs:   updateExprs,
			},
			Returning: &tree.NoReturningClause{},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&insert).String()}, true
	}

	// only bother translating inserts if there's an ON DUPLICATE KEY UPDATE clause, maybe revisit this later
	if len(stmt.OnDup) > 0 {
		tableName := translateTableName(table)

		var colList tree.NameList
		if len(stmt.Columns) > 0 {
			colList = make(tree.NameList, len(stmt.Columns))
			for i, col := range stmt.Columns {
				colList[i] = tree.Name(col.String())
			}
		}

		rows := rowsForInsert(stmt.Rows)

		onConflict := tree.OnConflict{
			Exprs:   convertUpdateExprs(sqlparser.AssignmentExprs(stmt.OnDup)),
			Columns: tree.NameList{tree.Name("fake")}, // column list ignored but must be present for valid syntax
		}

		insert := tree.Insert{
			Table:      tableName,
			Columns:    colList,
			Rows:       rows,
			OnConflict: &onConflict,
			Returning:  &tree.NoReturningClause{},
		}

		ctx := formatNodeWithUnqualifiedTableNames(&insert)
		return []string{ctx.String()}, true
	} else if stmt.Ignore == "ignore " {
		tableName := translateTableName(table)

		var colList tree.NameList
		if len(stmt.Columns) > 0 {
			colList = make(tree.NameList, len(stmt.Columns))
			for i, col := range stmt.Columns {
				colList[i] = tree.Name(col.String())
			}
		}

		rows := rowsForInsert(stmt.Rows)

		onConflict := tree.OnConflict{
			Columns:   tree.NameList{tree.Name("fake")}, // column list ignored but must be present for valid syntax
			DoNothing: true,
		}

		insert := tree.Insert{
			Table:      tableName,
			Columns:    colList,
			Rows:       rows,
			OnConflict: &onConflict,
			Returning:  &tree.NoReturningClause{},
		}

		ctx := formatNodeWithUnqualifiedTableNames(&insert)
		return []string{ctx.String()}, true
	}

	// Plain INSERTs still need the qualified-name rewrite when they reference
	// a `db.tbl` target so the table gets the postgres `db.public.tbl` form.
	if containsQualifiedTableName(stmt) {
		return []string{formatNode(stmt)}, true
	}

	return nil, false
}

// translateTableName converts a vitess MySQL TableName into a postgres TableName.
// MySQL `db.tbl` maps to postgres `db.public.tbl`; bare `tbl` stays bare.
// We mark the catalog/schema as "explicit" only when the source had a db
// qualifier so the format callback in formatNodeWithUnqualifiedTableNames can
// strip the prefix for unqualified names.
//
// Known schema names (`information_schema`, `pg_catalog`, etc.) are not
// translated: MySQL exposes them as if they were databases, but postgres
// exposes them as schemas, and the same `<schema>.<tbl>` form works in both.
func translateTableName(table sqlparser.TableName) *tree.TableName {
	if table.DbQualifier.IsEmpty() {
		// Unqualified: just the table name; let the formatter omit the prefix.
		return tree.NewUnqualifiedTableName(tree.Name(table.Name.String()))
	}
	if isWellKnownSchema(table.DbQualifier.String()) {
		// `schema.tbl` in postgres terms — encode the qualifier as the schema,
		// not the catalog. ExplicitSchema/ExplicitCatalog handling differs in
		// the format callback below.
		tn := tree.MakeTableNameWithSchema(
			"",
			tree.Name(table.DbQualifier.String()),
			tree.Name(table.Name.String()),
		)
		return &tn
	}
	tn := tree.MakeTableNameWithSchema(
		tree.Name(table.DbQualifier.String()),
		"public",
		tree.Name(table.Name.String()),
	)
	return &tn
}

// isWellKnownSchema returns true if the given identifier is a postgres-style
// schema (rather than a user database). When MySQL writes `information_schema.t`
// the user intent is the system schema; postgres exposes it under the same name.
func isWellKnownSchema(name string) bool {
	switch strings.ToLower(name) {
	case "information_schema", "pg_catalog", "performance_schema", "mysql", "sys":
		return true
	}
	return false
}

func TableNameToUnresolvedObjectName(table sqlparser.TableName) *tree.UnresolvedObjectName {
	if table.DbQualifier.IsEmpty() {
		name, err := tree.NewUnresolvedObjectName(1, [3]string{table.Name.String(), "", ""}, 0)
		if err != nil {
			panic(err)
		}
		return name
	}
	// MySQL db.tbl → postgres db.public.tbl (3 parts).
	name, err := tree.NewUnresolvedObjectName(3, [3]string{
		table.Name.String(),
		"public",
		table.DbQualifier.String(),
	}, 0)
	if err != nil {
		panic(err)
	}
	return name
}

func convertUpdateExprs(exprs sqlparser.AssignmentExprs) tree.UpdateExprs {
	updateExprs := make(tree.UpdateExprs, len(exprs))
	for i, expr := range exprs {
		updateExprs[i] = &tree.UpdateExpr{
			Names: tree.NameList{tree.Name(expr.Name.String())},
			Expr:  convertExpr(expr.Expr),
		}
	}
	return updateExprs
}

func rowsForInsert(rows sqlparser.InsertRows) *tree.Select {
	switch rows := rows.(type) {
	case sqlparser.Values:
		return &tree.Select{
			Select: &tree.ValuesClause{
				Rows: insertValuesToExprs(rows),
			},
		}
	case *sqlparser.Select:
		return &tree.Select{
			Select: convertSelect(rows),
		}
	case *sqlparser.ParenSelect:
		return &tree.Select{
			Select: &tree.ParenSelect{
				Select: convertParentSelect(rows.Select),
			},
		}
	case *sqlparser.AliasedValues:
		return &tree.Select{
			Select: &tree.ValuesClause{
				Rows: insertValuesToExprs(rows.Values),
			},
		}
	case *sqlparser.SetOp:
		return &tree.Select{
			Select: convertSelectStatement(rows),
		}
	default:
		panic(fmt.Sprintf("unhandled type: %T", rows))
	}
}

func convertParentSelect(statement sqlparser.SelectStatement) *tree.Select {
	switch statement := statement.(type) {
	case *sqlparser.Select:
		sel := convertSelect(statement)
		return &tree.Select{
			Select: sel,
		}
	default:
		panic(fmt.Sprintf("unhandled type: %T", statement))
	}
}

func convertSelect(sel *sqlparser.Select) *tree.SelectClause {
	return &tree.SelectClause{
		Distinct: sel.QueryOpts.Distinct,
		Exprs:    convertSelectExprs(sel.SelectExprs),
		From:     convertFrom(sel.From),
		Where:    convertWhere(sel.Where),
		GroupBy:  convertGroupBy(sel.GroupBy),
		Having:   convertHaving(sel.Having),
	}
}

func convertSelectStatement(sel sqlparser.SelectStatement) tree.SelectStatement {
	switch sel := sel.(type) {
	case *sqlparser.Select:
		return convertSelect(sel)
	case *sqlparser.SetOp:
		return convertSetOp(sel)
	default:
		panic(fmt.Sprintf("unhandled type: %T", sel))
	}
}

func convertSetOp(sel *sqlparser.SetOp) tree.SelectStatement {
	switch sel.Type {
	case sqlparser.UnionStr:
		left := convertSelectStatement(sel.Left)
		right := convertSelectStatement(sel.Right)
		return &tree.UnionClause{
			Type:  tree.UnionOp,
			Left:  selectFromSelectClause(left.(*tree.SelectClause)),
			Right: selectFromSelectClause(right.(*tree.SelectClause)),
		}
	default:
		panic(fmt.Sprintf("unhandled type: %s", sel.Type))
	}
}

func selectFromSelectClause(clause *tree.SelectClause) *tree.Select {
	return &tree.Select{
		Select: clause,
	}
}

func convertHaving(having *sqlparser.Where) *tree.Where {
	return convertWhere(having)
}

func convertGroupBy(groupBy sqlparser.GroupBy) tree.GroupBy {
	return convertExprs(sqlparser.Exprs(groupBy))
}

func convertWhere(where *sqlparser.Where) *tree.Where {
	if where == nil {
		return nil
	}
	return &tree.Where{
		Type: tree.AstWhere,
		Expr: convertExpr(where.Expr),
	}
}

func convertFrom(from sqlparser.TableExprs) tree.From {
	tables := make(tree.TableExprs, len(from))

	for i, table := range from {
		tables[i] = convertTableExpr(table)
	}
	return tree.From{
		Tables: tables,
	}
}

func convertTableExpr(table sqlparser.TableExpr) tree.TableExpr {
	switch table := table.(type) {
	case *sqlparser.AliasedTableExpr:
		switch tableExpr := table.Expr.(type) {
		case sqlparser.TableName:
			return &tree.AliasedTableExpr{
				Expr: translateTableName(tableExpr),
				As: tree.AliasClause{
					Alias: tree.Name(table.As.String()),
				},
			}
		default:
			panic(fmt.Sprintf("unhandled type: %T", table))
		}
	default:
		panic(fmt.Sprintf("unhandled type: %T", table))
	}
}

func convertSelectExprs(exprs sqlparser.SelectExprs) tree.SelectExprs {
	es := make(tree.SelectExprs, len(exprs))
	for i, expr := range exprs {
		es[i] = convertSelectExpr(expr)
	}
	return es
}

func insertValuesToExprs(values sqlparser.Values) []tree.Exprs {
	exprs := make([]tree.Exprs, len(values))
	for i, row := range values {
		exprs[i] = make(tree.Exprs, len(row))
		for j, val := range row {
			exprs[i][j] = convertValue(val)
		}
	}
	return exprs
}

func convertValue(val sqlparser.Expr) tree.Expr {
	switch val := val.(type) {
	case *sqlparser.SQLVal:
		return convertSQLVal(val)
	case *sqlparser.NullVal:
		return tree.DNull
	case *sqlparser.FuncExpr:
		return convertFuncExpr(val)
	default:
		panic(fmt.Sprintf("unhandled type: %T", val))
	}
}

func convertFuncExpr(val *sqlparser.FuncExpr) tree.Expr {
	fnName := tree.NewUnresolvedName(val.Name.String())
	exprs := make(tree.Exprs, len(val.Exprs))

	for i, expr := range val.Exprs {
		e := convertSelectExpr(expr)
		exprs[i] = e.Expr
	}
	return &tree.FuncExpr{
		Func: tree.ResolvableFunctionReference{
			FunctionReference: fnName,
		},
		Exprs: exprs,
	}
}

func convertSelectExpr(expr sqlparser.SelectExpr) tree.SelectExpr {
	switch val := expr.(type) {
	case *sqlparser.AliasedExpr:
		e := convertExpr(val.Expr)
		return tree.SelectExpr{
			Expr: e,
			As:   tree.UnrestrictedName(val.As.String()),
		}
	case *sqlparser.StarExpr:
		return tree.SelectExpr{
			Expr: tree.StarExpr(),
		}
	default:
		panic(fmt.Sprintf("unhandled type: %T", val))
	}
}

func convertExprs(exprs sqlparser.Exprs) []tree.Expr {
	es := make([]tree.Expr, len(exprs))
	for i, expr := range exprs {
		es[i] = convertExpr(expr)
	}
	return es
}

func convertExpr(expr sqlparser.Expr) tree.Expr {
	switch val := expr.(type) {
	case nil:
		return nil
	case *sqlparser.SQLVal:
		return convertSQLVal(val)
	case *sqlparser.ColName:
		// Translate MySQL column references into postgres equivalents.
		// MySQL uses a single qualifier for the database; postgres uses two
		// (catalog + schema), with `public` as the conventional default
		// schema when MySQL only gave a db name.
		//   `col`                        -> col
		//   `tbl.col`                    -> tbl.col
		//   `db.tbl.col`                 -> db.public.tbl.col
		//   `information_schema.tbl.col` -> information_schema.tbl.col
		if !val.Qualifier.DbQualifier.IsEmpty() {
			if isWellKnownSchema(val.Qualifier.DbQualifier.String()) {
				return tree.NewUnresolvedName(
					val.Qualifier.DbQualifier.String(),
					val.Qualifier.Name.String(),
					val.Name.String(),
				)
			}
			return tree.NewUnresolvedName(
				val.Qualifier.DbQualifier.String(),
				"public",
				val.Qualifier.Name.String(),
				val.Name.String(),
			)
		}
		if !val.Qualifier.Name.IsEmpty() {
			return tree.NewUnresolvedName(val.Qualifier.Name.String(), val.Name.String())
		}
		return tree.NewUnresolvedName(val.Name.String())
	case *sqlparser.FuncExpr:
		return convertFuncExpr(val)
	case *sqlparser.ValuesFuncExpr:
		return tree.NewStrVal(val.Name.String())
	case *sqlparser.BinaryExpr:
		return convertBinaryExpr(val)
	case *sqlparser.ComparisonExpr:
		return convertComparisonExpr(val)
	case *sqlparser.AndExpr:
		return &tree.AndExpr{
			Left:  convertExpr(val.Left),
			Right: convertExpr(val.Right),
		}
	case *sqlparser.OrExpr:
		return &tree.OrExpr{
			Left:  convertExpr(val.Left),
			Right: convertExpr(val.Right),
		}
	case *sqlparser.NotExpr:
		return &tree.NotExpr{
			Expr: convertExpr(val.Expr),
		}
	case *sqlparser.IsExpr:
		return convertIsExpr(val)
	case *sqlparser.UnaryExpr:
		return convertUnaryExpr(val)
	case *sqlparser.Subquery:
		return convertSubquery(val)
	case *sqlparser.ParenExpr:
		return &tree.ParenExpr{Expr: convertExpr(val.Expr)}
	case sqlparser.ValTuple:
		return convertValTuple(val)
	case *sqlparser.NullVal:
		return tree.DNull
	case sqlparser.BoolVal:
		boolVal := tree.DBool(bool(val))
		return &boolVal
	default:
		panic(fmt.Sprintf("unhandled type: %T", val))
	}
}

func convertIsExpr(val *sqlparser.IsExpr) tree.Expr {
	inner := convertExpr(val.Expr)
	switch val.Operator {
	case sqlparser.IsNullStr:
		return &tree.IsNullExpr{Expr: inner}
	case sqlparser.IsNotNullStr:
		return &tree.IsNotNullExpr{Expr: inner}
	case sqlparser.IsTrueStr:
		boolVal := tree.DBool(true)
		return &tree.ComparisonExpr{Operator: tree.IsNotDistinctFrom, Left: inner, Right: &boolVal}
	case sqlparser.IsNotTrueStr:
		boolVal := tree.DBool(true)
		return &tree.ComparisonExpr{Operator: tree.IsDistinctFrom, Left: inner, Right: &boolVal}
	case sqlparser.IsFalseStr:
		boolVal := tree.DBool(false)
		return &tree.ComparisonExpr{Operator: tree.IsNotDistinctFrom, Left: inner, Right: &boolVal}
	case sqlparser.IsNotFalseStr:
		boolVal := tree.DBool(false)
		return &tree.ComparisonExpr{Operator: tree.IsDistinctFrom, Left: inner, Right: &boolVal}
	default:
		panic(fmt.Sprintf("unhandled IS operator: %s", val.Operator))
	}
}

func convertUnaryExpr(val *sqlparser.UnaryExpr) tree.Expr {
	switch val.Operator {
	case sqlparser.UMinusStr:
		return &tree.UnaryExpr{Operator: tree.UnaryMinus, Expr: convertExpr(val.Expr)}
	case sqlparser.UPlusStr:
		// Unary plus is a no-op in both MySQL and Postgres
		return convertExpr(val.Expr)
	case sqlparser.TildaStr:
		return &tree.UnaryExpr{Operator: tree.UnaryComplement, Expr: convertExpr(val.Expr)}
	case sqlparser.BangStr:
		return &tree.NotExpr{Expr: convertExpr(val.Expr)}
	default:
		panic(fmt.Sprintf("unhandled unary operator: %s", val.Operator))
	}
}

func convertValTuple(val sqlparser.ValTuple) tree.Expr {
	exprs := make([]tree.Expr, len(val))
	for i, expr := range val {
		exprs[i] = convertExpr(expr)
	}
	return &tree.Tuple{Exprs: exprs}
}

func convertSubquery(val *sqlparser.Subquery) tree.Expr {
	return &tree.Subquery{
		Select: &tree.ParenSelect{
			// TODO: order by, limit
			Select: &tree.Select{
				Select: convertSelectStatement(val.Select),
			},
		},
	}
}

func convertComparisonExpr(val *sqlparser.ComparisonExpr) tree.Expr {
	var op tree.ComparisonOperator
	switch val.Operator {
	case sqlparser.EqualStr:
		op = tree.EQ
	case sqlparser.LessThanStr:
		op = tree.LT
	case sqlparser.LessEqualStr:
		op = tree.LE
	case sqlparser.GreaterThanStr:
		op = tree.GT
	case sqlparser.GreaterEqualStr:
		op = tree.GE
	case sqlparser.NotEqualStr:
		op = tree.NE
	case sqlparser.NullSafeEqualStr:
		// MySQL's <=> is null-safe equality; Postgres equivalent is IS NOT DISTINCT FROM
		op = tree.IsNotDistinctFrom
	case sqlparser.InStr:
		op = tree.In
	case sqlparser.NotInStr:
		op = tree.NotIn
	case sqlparser.LikeStr:
		op = tree.Like
	case sqlparser.NotLikeStr:
		op = tree.NotLike
	case sqlparser.RegexpStr:
		op = tree.RegMatch
	case sqlparser.NotRegexpStr:
		op = tree.NotRegMatch
	default:
		panic(fmt.Sprintf("unhandled operator: %s", val.Operator))
	}

	return &tree.ComparisonExpr{
		Operator: op,
		Left:     convertExpr(val.Left),
		Right:    convertExpr(val.Right),
		// Fn:       nil,
	}
}

func convertBinaryExpr(val *sqlparser.BinaryExpr) tree.Expr {
	var op tree.BinaryOperator
	switch val.Operator {
	case sqlparser.BitAndStr:
		op = tree.Bitand
	case sqlparser.BitOrStr:
		op = tree.Bitor
	case sqlparser.BitXorStr:
		op = tree.Bitxor
	case sqlparser.PlusStr:
		op = tree.Plus
	case sqlparser.MinusStr:
		op = tree.Minus
	case sqlparser.MultStr:
		op = tree.Mult
	case sqlparser.DivStr:
		op = tree.Div
	case sqlparser.IntDivStr:
		op = tree.FloorDiv
	case sqlparser.ModStr:
		op = tree.Mod
	case sqlparser.ShiftLeftStr:
		op = tree.LShift
	case sqlparser.ShiftRightStr:
		op = tree.RShift
	default:
		panic(fmt.Sprintf("unhandled operator: %s", val.Operator))
	}

	return &tree.BinaryExpr{
		Operator: op,
		Left:     convertExpr(val.Left),
		Right:    convertExpr(val.Right),
		// Fn:       nil,
	}
}

func convertSQLVal(val *sqlparser.SQLVal) tree.Expr {
	switch val.Type {
	case sqlparser.StrVal:
		return tree.NewStrVal(string(val.Val))
	case sqlparser.IntVal:
		i, err := strconv.Atoi(string(val.Val))
		if err != nil {
			panic(err)
		}
		return tree.NewDInt(tree.DInt(i))
	case sqlparser.FloatVal:
		f, err := strconv.ParseFloat(string(val.Val), 64)
		if err != nil {
			panic(err)
		}
		return tree.NewDFloat(tree.DFloat(f))
	case sqlparser.HexVal:
		return tree.NewStrVal(fmt.Sprintf("x'%s'", val.Val))
	case sqlparser.HexNum:
		return tree.NewStrVal(fmt.Sprintf("x'%s'", val.Val))
	default:
		panic(fmt.Sprintf("unhandled type: %v", val.Type))
	}
}

func transformDrop(query string, stmt *sqlparser.DDL) ([]string, bool) {
	// TODO
	return nil, false
}

func transformAlterTable(stmt *sqlparser.AlterTable) ([]string, bool) {
	var outputStmts []string
	for _, statement := range stmt.Statements {
		converted, ok := convertDdlStatement(statement)
		if !ok {
			return nil, false
		}
		outputStmts = append(outputStmts, converted...)
	}
	return outputStmts, true
}

func convertDdlStatement(statement *sqlparser.DDL) ([]string, bool) {
	switch statement.Action {
	case "alter":
		if statement.ColumnAction != "" {
			return convertColumnAlter(statement)
		}
		if statement.ConstraintAction != "" {
			return convertConstraintAlter(statement)
		}
		if statement.IndexSpec != nil {
			return convertIndexAlter(statement)
		}
		return nil, false
	case "rename":
		// ALTER TABLE ... RENAME TO ...  (table rename within an ALTER TABLE)
		if len(statement.FromTables) == 1 && len(statement.ToTables) == 1 {
			rename := &tree.RenameTable{
				Name:    TableNameToUnresolvedObjectName(statement.FromTables[0]),
				NewName: TableNameToUnresolvedObjectName(statement.ToTables[0]),
			}
			return []string{formatNodeWithUnqualifiedTableNames(rename).String()}, true
		}
		return nil, false
	default:
		return nil, false
	}
}

// convertColumnAlter handles ALTER TABLE ADD/DROP/MODIFY/CHANGE/RENAME COLUMN clauses.
func convertColumnAlter(statement *sqlparser.DDL) ([]string, bool) {
	tableName, err := tree.NewUnresolvedObjectName(1, [3]string{statement.Table.Name.String(), "", ""}, 0)
	if err != nil {
		panic(err)
	}

	switch statement.ColumnAction {
	case "add":
		if statement.TableSpec == nil || len(statement.TableSpec.Columns) != 1 {
			return nil, false
		}
		col := statement.TableSpec.Columns[0]
		// AUTO_INCREMENT in ADD COLUMN would require us to emit a CREATE SEQUENCE
		// statement too; skip for now to avoid silently producing a column without
		// a working default.
		if col.Type.Autoincrement {
			return nil, false
		}
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableAddColumn{
					IfNotExists: statement.IfNotExists,
					ColumnDef:   columnDefFromMySQL(col, ""),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true

	case "drop":
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableDropColumn{
					IfExists: statement.IfExists,
					Column:   tree.Name(statement.Column.String()),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true

	case "rename":
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableRenameColumn{
					Column:  tree.Name(statement.Column.String()),
					NewName: tree.Name(statement.ToColumn.String()),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true

	case "modify", "change":
		if statement.TableSpec == nil || len(statement.TableSpec.Columns) != 1 {
			return nil, false
		}
		col := statement.TableSpec.Columns[0]
		stmts := make([]string, 0)

		newType := convertTypeDef(col.Type)
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableAlterColumnType{
					Column: tree.Name(col.Name.String()),
					ToType: newType,
				},
			},
		}
		stmts = append(stmts, formatNodeWithUnqualifiedTableNames(&alter).String())

		if col.Type.NotNull {
			alter.Cmds = []tree.AlterTableCmd{
				&tree.AlterTableSetNotNull{Column: tree.Name(col.Name.String())},
			}
		} else {
			alter.Cmds = []tree.AlterTableCmd{
				&tree.AlterTableDropNotNull{Column: tree.Name(col.Name.String())},
			}
		}
		stmts = append(stmts, formatNodeWithUnqualifiedTableNames(&alter).String())

		// CHANGE may also rename the column.
		if statement.Column.String() != "" && statement.Column.String() != col.Name.String() {
			alter.Cmds = []tree.AlterTableCmd{
				&tree.AlterTableRenameColumn{
					Column:  tree.Name(statement.Column.String()),
					NewName: tree.Name(col.Name.String()),
				},
			}
			stmts = append(stmts, formatNodeWithUnqualifiedTableNames(&alter).String())
		}

		return stmts, true

	default:
		return nil, false
	}
}

// convertConstraintAlter handles ALTER TABLE ADD/DROP/RENAME CONSTRAINT clauses,
// covering foreign keys and check constraints.
func convertConstraintAlter(statement *sqlparser.DDL) ([]string, bool) {
	if statement.TableSpec == nil || len(statement.TableSpec.Constraints) == 0 {
		return nil, false
	}

	switch statement.ConstraintAction {
	case "add":
		// One constraint per ADD clause in the MySQL parse tree.
		c := statement.TableSpec.Constraints[0]
		tableName := *translateTableName(statement.Table)
		switch d := c.Details.(type) {
		case *sqlparser.ForeignKeyDefinition:
			return []string{createForeignKeyStatement(tableName, d)}, true
		case *sqlparser.CheckConstraintDefinition:
			return []string{createCheckConstraintStatement(tableName, d)}, true
		default:
			return nil, false
		}

	case "drop":
		// MySQL has separate DROP FOREIGN KEY / DROP CHECK syntax; postgres uses
		// generic DROP CONSTRAINT regardless of constraint kind.
		c := statement.TableSpec.Constraints[0]
		tableName, err := tree.NewUnresolvedObjectName(1, [3]string{statement.Table.Name.String(), "", ""}, 0)
		if err != nil {
			panic(err)
		}
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableDropConstraint{
					Constraint: tree.Name(c.Name),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true

	case "rename":
		// MySQL: ALTER TABLE t RENAME [CONSTRAINT] foo TO bar. Vitess models this
		// by stashing the two names in TableSpec.Constraints (first = old, second = new).
		if len(statement.TableSpec.Constraints) < 2 {
			return nil, false
		}
		tableName, err := tree.NewUnresolvedObjectName(1, [3]string{statement.Table.Name.String(), "", ""}, 0)
		if err != nil {
			panic(err)
		}
		alter := tree.AlterTable{
			Table: tableName,
			Cmds: []tree.AlterTableCmd{
				&tree.AlterTableRenameConstraint{
					Constraint: tree.Name(statement.TableSpec.Constraints[0].Name),
					NewName:    tree.Name(statement.TableSpec.Constraints[1].Name),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true

	default:
		return nil, false
	}
}

// convertIndexAlter handles ALTER TABLE ADD/DROP/RENAME INDEX clauses.
// Postgres expresses CREATE INDEX as a top-level statement (not an ALTER TABLE
// command), so we emit a CREATE INDEX statement when adding.
func convertIndexAlter(statement *sqlparser.DDL) ([]string, bool) {
	switch statement.IndexSpec.Action {
	case "drop":
		tableName := translateTableName(statement.Table)
		indexName := statement.IndexSpec.ToName.String()
		if statement.IndexSpec.Type == "primary" {
			indexName = "PRIMARY"
		}
		dropIndex := tree.DropIndex{
			IndexList: tree.TableIndexNames{
				{
					Table: *tableName,
					Index: tree.UnrestrictedName(indexName),
				},
			},
		}
		return []string{formatNodeWithUnqualifiedTableNames(&dropIndex).String()}, true

	case "create":
		// Bail out for index kinds the converter can't faithfully translate yet:
		// MySQL prefix-key syntax (col(N)) and expression-only fields (functional
		// indexes like `((LOWER(name)))`). Without bailing, we'd emit a
		// malformed CreateIndex tree that panics on Format.
		for _, f := range statement.IndexSpec.Fields {
			if f.Length != nil || f.Column.IsEmpty() {
				return nil, false
			}
		}

		// MySQL's ALTER TABLE ... ADD INDEX/UNIQUE/PRIMARY KEY. The PRIMARY KEY
		// variant has no postgres ALTER form that survives the existing harness
		// (Doltgres has no DROP PRIMARY KEY either), so leave it alone.
		if statement.IndexSpec.Type == "primary" {
			tableName, err := tree.NewUnresolvedObjectName(1, [3]string{statement.Table.Name.String(), "", ""}, 0)
			if err != nil {
				panic(err)
			}
			cols := make(tree.IndexElemList, len(statement.IndexSpec.Fields))
			for i, f := range statement.IndexSpec.Fields {
				cols[i] = tree.IndexElem{Column: tree.Name(f.Column.String())}
			}
			alter := tree.AlterTable{
				Table: tableName,
				Cmds: []tree.AlterTableCmd{
					&tree.AlterTableAddConstraint{
						ConstraintDef: &tree.UniqueConstraintTableDef{
							PrimaryKey: true,
							IndexTableDef: tree.IndexTableDef{
								Columns: cols,
							},
						},
					},
				},
			}
			return []string{formatNodeWithUnqualifiedTableNames(&alter).String()}, true
		}

		// Skip fulltext/spatial/vector — Postgres has no direct equivalent.
		if statement.IndexSpec.Type == "fulltext" || statement.IndexSpec.Type == "spatial" || statement.IndexSpec.Type == "vector" {
			return nil, false
		}

		cols := make(tree.IndexElemList, len(statement.IndexSpec.Fields))
		for i, f := range statement.IndexSpec.Fields {
			cols[i] = tree.IndexElem{
				Column:    tree.Name(f.Column.String()),
				Direction: tree.Ascending,
			}
		}
		createIndex := tree.CreateIndex{
			Name:    tree.Name(statement.IndexSpec.ToName.String()),
			Table:   *translateTableName(statement.Table),
			Unique:  statement.IndexSpec.Type == "unique",
			Columns: cols,
		}
		return []string{formatNodeWithUnqualifiedTableNames(&createIndex).String()}, true

	case "rename":
		tableName := translateTableName(statement.Table)
		from := statement.IndexSpec.FromName.String()
		to := statement.IndexSpec.ToName.String()
		renameIndex := tree.RenameIndex{
			Index: &tree.TableIndexName{
				Table: *tableName,
				Index: tree.UnrestrictedName(from),
			},
			NewName: tree.UnrestrictedName(to),
		}
		return []string{formatNodeWithUnqualifiedTableNames(&renameIndex).String()}, true

	default:
		return nil, false
	}
}

// transformSelect converts a MySQL SELECT statement to a postgres-compatible SELECT statement.
// This is a very broad surface area, so we do this very selectively
func transformSelect(stmt *sqlparser.Select) ([]string, bool) {
	if !shouldRewriteSelect(stmt) {
		return nil, false
	}
	return []string{formatNode(stmt)}, true
}

func shouldRewriteSelect(stmt *sqlparser.Select) bool {
	return containsUserVars(stmt) ||
		containsBinaryConversion(stmt) ||
		containsQualifiedTableName(stmt)
}

// containsQualifiedTableName reports whether the given node references a
// MySQL `db.tbl` identifier whose qualifier is a real user database (not a
// well-known postgres schema). Such queries need to be re-formatted so the
// converter can insert the postgres `public` schema between catalog and
// table. Queries that only reference `information_schema` etc. are left
// untouched.
func containsQualifiedTableName(stmt sqlparser.SQLNode) bool {
	found := false
	walker := func(node sqlparser.SQLNode) (bool, error) {
		switch n := node.(type) {
		case sqlparser.TableName:
			if !n.DbQualifier.IsEmpty() && !isWellKnownSchema(n.DbQualifier.String()) {
				found = true
				return false, nil
			}
		case *sqlparser.ColName:
			if !n.Qualifier.DbQualifier.IsEmpty() && !isWellKnownSchema(n.Qualifier.DbQualifier.String()) {
				found = true
				return false, nil
			}
		}
		return true, nil
	}
	sqlparser.Walk(walker, stmt)
	return found
}

func containsBinaryConversion(stmt *sqlparser.Select) bool {
	foundBinaryConversionExpr := false
	fn := func(node sqlparser.SQLNode) (bool, error) {
		switch node := node.(type) {
		case *sqlparser.BinaryExpr:
			if node.Operator == "binary " {
				foundBinaryConversionExpr = true
				return false, nil
			}
		case *sqlparser.UnaryExpr:
			if node.Operator == "binary " {
				foundBinaryConversionExpr = true
				return false, nil
			}
		}
		return true, nil
	}

	for _, sel := range stmt.SelectExprs {
		sqlparser.Walk(fn, sel)
	}

	if stmt.Where != nil {
		sqlparser.Walk(fn, stmt.Where)
	}

	return foundBinaryConversionExpr
}

func containsUserVars(stmt *sqlparser.Select) bool {
	foundUserVar := false
	detectUserVar := func(node sqlparser.SQLNode) (bool, error) {
		switch node := node.(type) {
		case *sqlparser.ColName:
			if strings.HasPrefix(node.Name.String(), "@") && !strings.HasPrefix(node.Name.String(), "@@") {
				foundUserVar = true
				return false, nil
			}
		}
		return true, nil
	}
	// Walk the whole statement so user variables get detected anywhere they
	// appear — SELECT expressions, WHERE, GROUP BY, ORDER BY, and crucially
	// FROM ... AS OF expressions (e.g. `FROM tbl AS OF @rev1`).
	sqlparser.Walk(detectUserVar, stmt)
	return foundUserVar
}

func transformSet(stmt *sqlparser.Set) ([]string, bool) {
	var queries []string

	// the semantics aren't quite the same, but setting autocommit to false is the same as beginning a transaction
	// (for most scripts). Setting autocommit to true is a no-op.
	if len(stmt.Exprs) == 1 && strings.ToLower(stmt.Exprs[0].Name.String()) == "autocommit" {
		exprStr := strings.ToLower(formatNode(stmt.Exprs[0].Expr))
		if exprStr == "0" || exprStr == "off" || exprStr == "'off'" || exprStr == "false" {
			queries = append(queries, "START TRANSACTION")
			return queries, true
		} else {
			return []string{""}, true
		}
	}

	for _, expr := range stmt.Exprs {
		if expr.Scope == sqlparser.GlobalStr {
			queries = append(queries, fmt.Sprintf("SET GLOBAL %s = %s", expr.Name, expr.Expr))
		} else if expr.Scope == "user" {
			queries = append(queries, fmt.Sprintf("SET doltgres_enginetest.%s = %s", expr.Name, formatNode(expr.Expr)))
		} else {
			queries = append(queries, fmt.Sprintf("SET %s = %s", expr.Name, expr.Expr))
		}
	}

	return queries, true
}

func formatNode(node sqlparser.SQLNode) string {
	buf := sqlparser.NewTrackedBuffer(PostgresNodeFormatter)
	node.Format(buf)
	return buf.String()
}

func PostgresNodeFormatter(buf *sqlparser.TrackedBuffer, node sqlparser.SQLNode) {
	switch node := node.(type) {
	case sqlparser.ColIdent:
		// MySQL `@var` and `@@var` references map to postgres' current_setting().
		// All other identifiers go through tree.AutoQuoteIdent so that names
		// containing special characters or matching a reserved keyword come out
		// double-quoted (postgres' quoting style) rather than backticked
		// (vitess' default formatID behavior).
		switch {
		case strings.HasPrefix(node.String(), "@@"):
			buf.Myprintf("current_setting('.%s')", strings.TrimLeft(node.String(), "@"))
		case strings.HasPrefix(node.String(), "@"):
			buf.Myprintf("current_setting('doltgres_enginetest.%s')", strings.TrimLeft(node.String(), "@"))
		default:
			buf.WriteString(AutoQuoteIdent(node.Lowered()))
		}
	case sqlparser.TableIdent:
		// Same auto-quoting story as ColIdent; vitess' formatID would
		// otherwise wrap special-character identifiers in backticks.
		buf.WriteString(AutoQuoteIdent(node.String()))
	case sqlparser.TableName:
		// MySQL `db.tbl` → postgres `db.public.tbl`. MySQL has no schema
		// concept, so we insert `public` (postgres' default schema) between
		// the catalog and table when a database qualifier is present —
		// except for well-known schemas like `information_schema`, where the
		// MySQL qualifier already lines up with a postgres schema.
		switch {
		case node.DbQualifier.IsEmpty():
			buf.Myprintf("%v", node.Name)
		case isWellKnownSchema(node.DbQualifier.String()):
			buf.Myprintf("%v.%v", node.DbQualifier, node.Name)
		default:
			buf.Myprintf("%v.public.%v", node.DbQualifier, node.Name)
		}
	case *sqlparser.ColName:
		// Same translation for qualified column references: `db.tbl.col` →
		// `db.public.tbl.col`, leaving `information_schema.tbl.col` alone.
		switch {
		case node.Qualifier.DbQualifier.IsEmpty() && node.Qualifier.Name.IsEmpty():
			buf.Myprintf("%v", node.Name)
		case node.Qualifier.DbQualifier.IsEmpty():
			buf.Myprintf("%v.%v", node.Qualifier.Name, node.Name)
		case isWellKnownSchema(node.Qualifier.DbQualifier.String()):
			buf.Myprintf("%v.%v.%v", node.Qualifier.DbQualifier, node.Qualifier.Name, node.Name)
		default:
			buf.Myprintf("%v.public.%v.%v", node.Qualifier.DbQualifier, node.Qualifier.Name, node.Name)
		}
	case *sqlparser.UnaryExpr:
		if node.Operator == "binary " {
			buf.Myprintf("%v::text::bytea", node.Expr)
		} else {
			buf.Myprintf("%v", node)
		}
	case *sqlparser.Limit:
		if node == nil {
			return
		}
		buf.Myprintf(" limit %v", node.Rowcount)
		if node.Offset != nil {
			buf.Myprintf(" offset %v", node.Offset)
		}
	default:
		node.Format(buf)
	}
}

var sequenceNum int

// columnDefFromMySQL converts a parsed MySQL column definition into a Postgres
// ColumnTableDef. The optional sequenceName argument is used as the source for
// AUTO_INCREMENT columns; passing "" means callers don't want auto-increment
// handled (e.g. ALTER TABLE ADD COLUMN where we don't currently emit a
// CREATE SEQUENCE alongside).
func columnDefFromMySQL(col *sqlparser.ColumnDefinition, sequenceName string) *tree.ColumnTableDef {
	defVal := convertExpr(col.Type.Default)

	if col.Type.Autoincrement && sequenceName != "" {
		defVal = &tree.FuncExpr{
			Func: tree.WrapFunction("nextval"),
			Exprs: []tree.Expr{
				tree.NewStrVal(sequenceName),
			},
		}
	}

	return &tree.ColumnTableDef{
		Name:      tree.Name(col.Name.String()),
		Type:      convertTypeDef(col.Type),
		Collation: "", // TODO
		Nullable: struct {
			Nullability    tree.Nullability
			ConstraintName tree.Name
		}{
			Nullability: convertNullability(col.Type),
		},
		PrimaryKey: struct {
			IsPrimaryKey bool
		}{
			IsPrimaryKey: col.Type.KeyOpt == 1, // TODO: unexported const
		},
		Unique:               col.Type.KeyOpt == 3, // TODO: unexported const
		UniqueConstraintName: "",                   // TODO
		DefaultExpr: struct {
			Expr           tree.Expr
			ConstraintName tree.Name
		}{
			Expr:           defVal,
			ConstraintName: "", // TODO
		},
		CheckExprs: nil, // TODO
	}
}

func transformCreateTable(stmt *sqlparser.DDL) ([]string, bool) {
	if stmt.TableSpec == nil {
		return nil, false
	}

	createTable := tree.CreateTable{
		IfNotExists: stmt.IfNotExists,
		Table:       *translateTableName(stmt.Table),
	}

	var queries []string
	var autoIncColumn string
	for _, col := range stmt.TableSpec.Columns {
		seqName := ""
		if col.Type.Autoincrement {
			autoIncColumn = col.Name.String()
			seqName = fmt.Sprintf("seq_%d", sequenceNum)
		}
		createTable.Defs = append(createTable.Defs, columnDefFromMySQL(col, seqName))
	}

	// convert any primary key indexes
	if len(stmt.TableSpec.Indexes) > 0 {
		for _, index := range stmt.TableSpec.Indexes {
			if !index.Info.Primary {
				continue
			}

			indexFields := make(tree.IndexElemList, len(index.Fields))
			for i, col := range index.Fields {
				colName := col.Column.String()
				indexFields[i] = tree.IndexElem{
					Column: tree.Name(colName),
				}
			}

			indexDef := &tree.UniqueConstraintTableDef{
				PrimaryKey: true,
				IndexTableDef: tree.IndexTableDef{
					Columns: indexFields,
				},
			}

			createTable.Defs = append(createTable.Defs, indexDef)
		}
	}

	if autoIncColumn != "" {
		queries = append(queries, fmt.Sprintf("CREATE SEQUENCE seq_%d", sequenceNum))
		sequenceNum++
	}

	ctx := formatNodeWithUnqualifiedTableNames(&createTable)
	query := ctx.String()

	// this is a very odd quirk for only the char type, not sure why the postgres parser does this but it doesn't
	// parse in a CREATE TABLE statement
	query = strings.ReplaceAll(query, `"char"`, `char`)
	queries = append(queries, query)

	// If there are additional (non-primary key) indexes defined, each one gets its own additional statement
	if len(stmt.TableSpec.Indexes) > 0 {
		for _, index := range stmt.TableSpec.Indexes {
			if index.Info.Primary {
				continue
			}

			createIndex := tree.CreateIndex{
				Name:    tree.Name(index.Info.Name.String()),
				Table:   *translateTableName(stmt.Table),
				Unique:  index.Info.Unique,
				Columns: make(tree.IndexElemList, len(index.Fields)),
			}

			for i, col := range index.Fields {
				createIndex.Columns[i] = tree.IndexElem{
					Column:    tree.Name(col.Column.String()),
					Direction: tree.Ascending,
				}
			}

			ctx := formatNodeWithUnqualifiedTableNames(&createIndex)

			queries = append(queries, ctx.String())
		}
	}

	// convert constraints into separate statements as well
	for _, c := range stmt.TableSpec.Constraints {
		switch c := c.Details.(type) {
		case *sqlparser.ForeignKeyDefinition:
			queries = append(queries, createForeignKeyStatement(createTable.Table, c))
		case *sqlparser.CheckConstraintDefinition:
			queries = append(queries, createCheckConstraintStatement(createTable.Table, c))
		default:
			// do nothing, unsupported
		}
	}

	return queries, true
}

func createCheckConstraintStatement(table tree.TableName, c *sqlparser.CheckConstraintDefinition) string {
	name, err := tree.NewUnresolvedObjectName(1, [3]string{table.Table(), "", ""}, 0)
	if err != nil {
		panic(err)
	}

	alter := tree.AlterTable{
		Table: name,
	}

	alter.Cmds = append(alter.Cmds, &tree.AlterTableAddConstraint{
		ConstraintDef: &tree.CheckConstraintTableDef{
			Expr: convertExpr(c.Expr),
		},
	})

	ctx := formatNodeWithUnqualifiedTableNames(&alter)
	return ctx.String()
}

func createForeignKeyStatement(table tree.TableName, c *sqlparser.ForeignKeyDefinition) string {
	name, err := tree.NewUnresolvedObjectName(1, [3]string{table.Table(), "", ""}, 0)
	if err != nil {
		panic(err)
	}

	alter := tree.AlterTable{
		Table: name,
	}

	var fromCols, toCols tree.NameList
	for _, col := range c.Source {
		fromCols = append(fromCols, tree.Name(col.String()))
	}
	for _, col := range c.ReferencedColumns {
		toCols = append(toCols, tree.Name(col.String()))
	}

	onDelete := translateRefAction(c.OnDelete)
	onUpdate := translateRefAction(c.OnUpdate)

	alter.Cmds = append(alter.Cmds, &tree.AlterTableAddConstraint{
		ConstraintDef: &tree.ForeignKeyConstraintTableDef{
			FromCols: fromCols,
			Table:    *translateTableName(c.ReferencedTable),
			ToCols:   toCols,
			Actions: tree.ReferenceActions{
				Delete: onDelete,
				Update: onUpdate,
			},
		},
	})

	ctx := formatNodeWithUnqualifiedTableNames(&alter)
	return ctx.String()
}

func translateRefAction(action sqlparser.ReferenceAction) tree.RefAction {
	switch action {
	case sqlparser.Cascade:
		return tree.RefAction{
			Action: tree.Cascade,
		}
	case sqlparser.SetNull:
		return tree.RefAction{
			Action: tree.SetNull,
		}
	case sqlparser.NoAction:
		return tree.RefAction{
			Action: tree.NoAction,
		}
	case sqlparser.Restrict:
		return tree.RefAction{
			Action: tree.Restrict,
		}
	case sqlparser.SetDefault:
		return tree.RefAction{
			Action: tree.SetDefault,
		}
	case sqlparser.DefaultAction:
		return tree.RefAction{
			Action: tree.Restrict, // is this correct?
		}
	default:
		panic(fmt.Sprintf("unhandled on delete action: %v", action))
	}
}

// formatNodeWithUnqualifiedTableNames formats a node and, by default, prints
// table names unqualified. The default postgres formatter always emits the
// fully-qualified `catalog.schema.table` form, which is too noisy when we
// haven't actually been given a qualifier. We do, however, want to preserve
// qualifiers when the source MySQL query had one — see translateTableName for
// the rules.
func formatNodeWithUnqualifiedTableNames(n tree.NodeFormatter) *tree.FmtCtx {
	ctx := tree.NewFmtCtx(tree.FmtSimple)
	ctx.SetReformatTableNames(func(ctx *tree.FmtCtx, tn *tree.TableName) {
		switch {
		case tn.ExplicitCatalog && tn.CatalogName != "":
			ctx.FormatNode(&tn.CatalogName)
			ctx.WriteByte('.')
			schema := tn.SchemaName
			if schema == "" {
				schema = "public"
			}
			ctx.FormatNode(&schema)
			ctx.WriteByte('.')
		case tn.ExplicitSchema && tn.SchemaName != "":
			// Schema-only qualifier (e.g. `information_schema.tbl`); the catalog
			// is left implicit so postgres uses the current database.
			ctx.FormatNode(&tn.SchemaName)
			ctx.WriteByte('.')
		}
		ctx.FormatNode(&tn.ObjectName)
	})
	ctx.FormatNode(n)
	return ctx
}

func convertNullability(typ sqlparser.ColumnType) tree.Nullability {
	if typ.NotNull {
		return tree.NotNull
	}
	if typ.KeyOpt == 1 { // primary key, unexported constant
		return tree.NotNull
	}

	return tree.Null
}

func convertTypeDef(columnType sqlparser.ColumnType) tree.ResolvableTypeReference {
	switch strings.ToLower(columnType.Type) {
	case "int", "mediumint", "integer":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.IntFamily,
				Width:  32,
				Oid:    oid.T_int4,
			},
		}
	case "tinyint", "smallint", "bool":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.IntFamily,
				Width:  16,
				Oid:    oid.T_int2,
			},
		}
	case "bigint":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.IntFamily,
				Width:  64,
				Oid:    oid.T_int8,
			},
		}
	case "float", "real":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.FloatFamily,
				Width:  32,
				Oid:    oid.T_float4,
			},
		}
	case "double precision", "double":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.FloatFamily,
				Oid:    oid.T_float8,
			},
		}
	case "decimal":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.DecimalFamily,
				Oid:    oid.T_numeric,
			},
		}
	case "varchar":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.StringFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_varchar,
			},
		}
	case "char":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.StringFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_char,
			},
		}
	case "text", "tinytext", "mediumtext", "longtext":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.StringFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_text,
			},
		}
	case "blob", "binary", "varbinary", "tinyblob", "mediumblob", "longblob":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.BytesFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_bytea,
			},
		}
	case "datetime", "timestamp":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.TimestampFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_timestamp,
			},
		}
	case "date":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.DateFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_date,
			},
		}
	case "time":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.TimeFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_time,
			},
		}
	case "enum":
		panic(fmt.Sprintf("unhandled type: %s", columnType.Type))
	case "set":
		panic(fmt.Sprintf("unhandled type: %s", columnType.Type))
	case "bit":
		// Support for MySQL BIT type conversion (dolt#9641)
		// See: https://github.com/dolthub/dolt/issues/9641
		// Map BIT types to appropriately sized integers for MySQL compatibility
		width := int32FromSqlVal(columnType.Length)

		var intOid oid.Oid
		var intWidth int32
		if width <= 16 {
			intOid = oid.T_int2
			intWidth = 16
		} else if width <= 32 {
			intOid = oid.T_int4
			intWidth = 32
		} else {
			intOid = oid.T_int8
			intWidth = 64
		}

		return &types.T{
			InternalType: types.InternalType{
				Family: types.IntFamily,
				Width:  intWidth,
				Oid:    intOid,
			},
		}
	case "json":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.JsonFamily,
				Width:  int32FromSqlVal(columnType.Length),
				Oid:    oid.T_json,
			},
		}
	case "boolean":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.BoolFamily,
				Oid:    oid.T_bool,
			},
		}
	case "year":
		return &types.T{
			InternalType: types.InternalType{
				Family: types.IntFamily,
				Width:  16,
				Oid:    oid.T_int2,
			},
		}
	case "geometry", "point", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon", "geometrycollection":
		panic(fmt.Sprintf("unhandled type: %s", columnType.Type))
	default:
		panic(fmt.Sprintf("unhandled type: %s", columnType.Type))
	}
}

func int32FromSqlVal(v *sqlparser.SQLVal) int32 {
	if v == nil {
		return 0
	}

	i, err := strconv.Atoi(string(v.Val))
	if err != nil {
		return 0
	}
	return int32(i)
}

var doltProcedureCall = regexp.MustCompile(`(?i)CALL DOLT_(\w+)`)

func convertDoltProcedureCalls(query string) string {
	return doltProcedureCall.ReplaceAllString(query, "SELECT DOLT_$1")
}

// little state machine for turning MySQL quote characters into their postgres equivalents:
/*
               ┌───────────────────*─────────────────────────┐
               │                   ┌─*─┐                     *
               │               ┌───┴───▼──────┐         ┌────┴─────────┐
               │     ┌────"───►│ In double    │◄───"────┤End double    │
               │     │         │ quoted string│────"───►│quoted string?│
               │     │         └──────────────┘         └──────────────┘
               ├─────(──────────────────*───────────────────┐
      ┌─*──┐   ▼     │                                      *
      │    ├─────────┴┐            ┌─*─┐                    │
      └───►│ Not in   │        ┌───┴───▼─────┐          ┌───┴──────────┐
           │ string   ├───'───►│In single    │◄────'────┤End single    │
  ────────►└─────────┬┘        │quoted string│─────'───►│quoted string?│
  START        ▲     │         └─────────────┘          └──────────────┘
               └─────(──────────────────*───────────────────┐
                     │            ┌─*──┐                    *
                     │        ┌───┴────▼────┐           ┌───┴──────────┐
                     └───`───►│In backtick  │◄─────`────┤End backtick  │
                              │quoted string│──────`───►│quoted string?│
                              └─────────────┘           └──────────────┘
*/
type stringParserState byte

const (
	notInString stringParserState = iota
	inDoubleQuote
	maybeEndDoubleQuote
	inSingleQuote
	maybeEndSingleQuote
	inBackticks
	maybeEndBackticks
)

const singleQuote = '\''
const doubleQuote = '"'
const backtick = '`'
const backslash = '\\'

// normalizeStrings normalizes a query string to convert any MySQL syntax to Postgres syntax
func normalizeStrings(q string) string {
	state := notInString
	lastCharWasBackslash := false
	normalized := strings.Builder{}

	for _, c := range q {
		switch state {
		case notInString:
			switch c {
			case singleQuote:
				state = inSingleQuote
				normalized.WriteRune(singleQuote)
			case doubleQuote:
				state = inDoubleQuote
				normalized.WriteRune(singleQuote)
			case backtick:
				state = inBackticks
				normalized.WriteRune(doubleQuote)
			default:
				normalized.WriteRune(c)
			}
		case inDoubleQuote:
			switch c {
			case backslash:
				if lastCharWasBackslash {
					normalized.WriteRune(c)
				}
				lastCharWasBackslash = !lastCharWasBackslash
			case doubleQuote:
				if lastCharWasBackslash {
					normalized.WriteRune(c)
					lastCharWasBackslash = false
				} else {
					state = maybeEndDoubleQuote
				}
			case singleQuote:
				normalized.WriteRune(singleQuote)
				normalized.WriteRune(singleQuote)
				lastCharWasBackslash = false
			default:
				lastCharWasBackslash = false
				normalized.WriteRune(c)
			}
		case maybeEndDoubleQuote:
			switch c {
			case doubleQuote:
				state = inDoubleQuote
				normalized.WriteRune(doubleQuote)
			default:
				state = notInString
				normalized.WriteRune(singleQuote)
				normalized.WriteRune(c)
			}
		case inSingleQuote:
			switch c {
			case backslash:
				if lastCharWasBackslash {
					normalized.WriteRune(c)
				}
				lastCharWasBackslash = !lastCharWasBackslash
			case singleQuote:
				if lastCharWasBackslash {
					normalized.WriteRune(c)
					normalized.WriteRune(c)
					lastCharWasBackslash = false
				} else {
					state = maybeEndSingleQuote
				}
			default:
				lastCharWasBackslash = false
				normalized.WriteRune(c)
			}
		case maybeEndSingleQuote:
			switch c {
			case singleQuote:
				state = inSingleQuote
				normalized.WriteRune(singleQuote)
				normalized.WriteRune(singleQuote)
			default:
				state = notInString
				normalized.WriteRune(singleQuote)
				normalized.WriteRune(c)
			}
		case inBackticks:
			switch c {
			case backtick:
				state = maybeEndBackticks
			default:
				normalized.WriteRune(c)
			}
		case maybeEndBackticks:
			switch c {
			case backtick:
				state = inBackticks
				normalized.WriteRune(backtick)
			default:
				state = notInString
				normalized.WriteRune(doubleQuote)
				normalized.WriteRune(c)
			}
		default:
			panic("unknown state")
		}
	}

	// If reached the end of input unsure whether to unquote a string, do so now
	switch state {
	case maybeEndDoubleQuote:
		normalized.WriteRune(singleQuote)
	case maybeEndSingleQuote:
		normalized.WriteRune(singleQuote)
	case maybeEndBackticks:
		normalized.WriteRune(doubleQuote)
	default: // do nothing
	}

	return normalized.String()
}

// Test converting MySQL strings to Postgres strings
func TestNormalizeStrings(t *testing.T) {
	type test struct {
		input    string
		expected string
	}
	tests := []test{
		{
			input:    "SELECT \"foo\" FROM `bar`",
			expected: `SELECT 'foo' FROM "bar"`,
		},
		{
			input:    `SELECT "foo"`,
			expected: `SELECT 'foo'`,
		},
		{
			input:    `SELECT "fo\"o"`,
			expected: `SELECT 'fo"o'`,
		},
		{
			input:    `SELECT "fo\'o"`,
			expected: `SELECT 'fo''o'`,
		},
		{
			input:    `SELECT 'fo\'o'`,
			expected: `SELECT 'fo''o'`,
		},
		{
			input:    `SELECT 'fo\"o'`,
			expected: `SELECT 'fo"o'`,
		},
		{
			input:    `SELECT 'fo\\"o'`,
			expected: `SELECT 'fo\"o'`,
		},
		{
			input:    `SELECT 'fo\\\'o'`,
			expected: `SELECT 'fo\''o'`,
		},
		{
			input:    `SELECT "fo\\'o"`,
			expected: `SELECT 'fo\''o'`,
		},
		{
			input:    `SELECT "fo\\\"o"`,
			expected: `SELECT 'fo\"o'`,
		},
		{
			input:    "SELECT 'fo''o'",
			expected: `SELECT 'fo''o'`,
		},
		{
			input:    "SELECT 'fo''''o'",
			expected: `SELECT 'fo''''o'`,
		},
		{
			input:    `SELECT "fo'o"`,
			expected: `SELECT 'fo''o'`,
		},
		{
			input:    `SELECT "fo''o"`,
			expected: `SELECT 'fo''''o'`,
		},
		{
			input:    `SELECT "fo""o"`,
			expected: `SELECT 'fo"o'`,
		},
		{
			input:    `SELECT "fo""""o"`,
			expected: `SELECT 'fo""o'`,
		},
		{
			input:    `SELECT 'fo""o'`,
			expected: `SELECT 'fo""o'`,
		},
		{
			input:    "SELECT `foo` FROM `bar`",
			expected: `SELECT "foo" FROM "bar"`,
		},
		{
			input:    "SELECT 'foo' FROM `bar`",
			expected: `SELECT 'foo' FROM "bar"`,
		},
		{
			input:    "SELECT `f\"o'o` FROM `ba``r`",
			expected: "SELECT \"f\"o'o\" FROM \"ba`r\"",
		},
		{
			input:    "SELECT \"foo\" from `bar` where `bar`.`baz` = \"qux\"",
			expected: `SELECT 'foo' from "bar" where "bar"."baz" = 'qux'`,
		},
	}

	for _, test := range tests {
		t.Run(test.input, func(t *testing.T) {
			actual := normalizeStrings(test.input)
			require.Equal(t, test.expected, actual)
		})
	}
}

// TestPostgresQueryFormat is a utility function to test how postgres parses and formats queries
func TestPostgresQueryFormat(t *testing.T) {
	type test struct {
		input    string
		expected string
	}
	tests := []test{
		{
			input:    "CREATE TABLE foo (a INT primary key default nextval('myseq'))",
			expected: "CREATE TABLE foo (a INTEGER DEFAULT nextval('myseq') PRIMARY KEY)",
		},
	}

	for _, test := range tests {
		t.Run(test.input, func(t *testing.T) {
			s, err := parser.Parse(test.input)
			require.NoError(t, err)

			ctx := formatNodeWithUnqualifiedTableNames(s[0].AST)
			query := ctx.String()
			require.Equal(t, test.expected, query)
		})
	}
}

func TestConvertQuery(t *testing.T) {
	type test struct {
		input    string
		expected []string
		pattern  bool
	}
	tests := []test{
		{
			input:    "CREATE TABLE foo (a INT primary key)",
			expected: []string{"CREATE TABLE foo (a INTEGER NOT NULL PRIMARY KEY)"},
		},
		{
			input: "CREATE TABLE foo (a INT primary key, b int not null)",
			expected: []string{
				"CREATE TABLE foo (a INTEGER NOT NULL PRIMARY KEY, b INTEGER NOT NULL)",
			},
		},
		{
			input: "CREATE TABLE foo (a INT primary key, b int, key (b))",
			expected: []string{
				"CREATE TABLE foo (a INTEGER NOT NULL PRIMARY KEY, b INTEGER NULL)",
				"CREATE INDEX ON foo ( b ASC ) NULLS NOT DISTINCT ",
			},
		},
		{
			input: "CREATE TABLE test (pk BIGINT PRIMARY KEY AUTO_INCREMENT, v1 BIGINT);",
			// this is a pattern match because when run with other tests in the same process, the name of the sequence created is changed
			pattern: true,
			expected: []string{
				"CREATE SEQUENCE .+",
				"CREATE TABLE test \\(pk BIGINT NOT NULL DEFAULT nextval\\('.+'\\) PRIMARY KEY, v1 BIGINT NULL\\)"},
		},
		{
			input:    "CREATE TABLE foo (a INT, b int, primary key (b,a))",
			expected: []string{"CREATE TABLE foo (a INTEGER NULL, b INTEGER NULL, PRIMARY KEY (b, a))"},
		},
		{
			input: "CREATE TABLE foo (a INT primary key, b int, c int, key (c,b))",
			expected: []string{
				"CREATE TABLE foo (a INTEGER NOT NULL PRIMARY KEY, b INTEGER NULL, c INTEGER NULL)",
				"CREATE INDEX ON foo ( c ASC, b ASC ) NULLS NOT DISTINCT ",
			},
		},
		{
			input: "CREATE TABLE foo (a INT primary key, b int, c int not null, d int, key (c), key (b), key (b,c))",
			expected: []string{
				"CREATE TABLE foo (a INTEGER NOT NULL PRIMARY KEY, b INTEGER NULL, c INTEGER NOT NULL, d INTEGER NULL)",
				"CREATE INDEX ON foo ( c ASC ) NULLS NOT DISTINCT ",
				"CREATE INDEX ON foo ( b ASC ) NULLS NOT DISTINCT ",
				"CREATE INDEX ON foo ( b ASC, c ASC ) NULLS NOT DISTINCT ",
			},
		},
		{
			input:    "SET @@autocommit = 1",
			expected: []string{""},
		},
		{
			input:    "SET @@autocommit = 0",
			expected: []string{"START TRANSACTION"},
		},
		{
			input:    "SET @@autocommit = off",
			expected: []string{"START TRANSACTION"},
		},
		{
			input: "SET @@autocommit = 1, @@dolt_transaction_commit = off",
			expected: []string{
				"SET autocommit = 1",
				"SET dolt_transaction_commit = 'off'",
			},
		},
		{
			input: "INSERT INTO foo (a, b) VALUES (1, 2), (3, 4) on duplicate key update a = 5",
			expected: []string{
				"INSERT INTO foo(a, b) VALUES (1, 2), (3, 4) ON CONFLICT (fake) DO UPDATE SET a = 5",
			},
		},
		{
			input: "INSERT INTO foo VALUES (1, 2), (3, 4) on duplicate key update a = 5",
			expected: []string{
				"INSERT INTO foo VALUES (1, 2), (3, 4) ON CONFLICT (fake) DO UPDATE SET a = 5",
			},
		},
	}

	for _, test := range tests {
		t.Run(test.input, func(t *testing.T) {
			actual := convertQuery(test.input)
			if test.pattern {
				require.Equal(t, len(test.expected), len(actual))
				for i := range test.expected {
					require.Regexp(t, test.expected[i], actual[i])
				}
			} else {
				require.Equal(t, test.expected, actual)
			}
		})
	}
}

// TestBoolValSupport tests that query converter can handle boolean literals
// Related to issue #1708: Query converter crashes on boolean literals
func TestBoolValSupport(t *testing.T) {
	// This query contains boolean literals that should be converted properly
	// Before the fix: panics with "unhandled type: sqlparser.BoolVal"
	// After the fix: should convert successfully without panic
	result := convertQuery("CREATE TABLE test_table (id INT, archived BOOLEAN DEFAULT FALSE)")

	// Should not panic and should return converted query
	require.NotEmpty(t, result, "Query conversion should succeed and return converted SQL")
	require.Len(t, result, 1, "Should return exactly one converted statement")
	require.Contains(t, result[0], "CREATE TABLE", "Result should contain CREATE TABLE")
	require.Contains(t, result[0], "false", "Result should contain converted boolean literal")
}

// TestTranslateMysqlShowCreateTable covers the harness-side rewrite that
// turns the MySQL-format CREATE TABLE text in the test fixtures into the
// doltgres/postgres dialect, so SHOW CREATE TABLE assertions compare apples
// to apples.
func TestTranslateMysqlShowCreateTable(t *testing.T) {
	type test struct {
		name     string
		input    string
		expected string
	}
	tests := []test{
		{
			name: "basic types and PRIMARY KEY",
			input: "CREATE TABLE `a` (\n" +
				"  `pk` int NOT NULL,\n" +
				"  `c1` int,\n" +
				"  PRIMARY KEY (`pk`)\n" +
				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin",
			expected: "CREATE TABLE \"a\" (\n" +
				"  \"pk\" integer NOT NULL,\n" +
				"  \"c1\" integer,\n" +
				"  PRIMARY KEY (\"pk\")\n" +
				")",
		},
		{
			name: "non-unique KEY dropped; UNIQUE KEY becomes CONSTRAINT",
			input: "CREATE TABLE `tbl` (\n" +
				"  `a` int NOT NULL,\n" +
				"  PRIMARY KEY (`a`),\n" +
				"  KEY `tbl_bc` (`b`,`c`),\n" +
				"  UNIQUE KEY `tbl_c` (`c`)\n" +
				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin",
			expected: "CREATE TABLE \"tbl\" (\n" +
				"  \"a\" integer NOT NULL,\n" +
				"  PRIMARY KEY (\"a\"),\n" +
				"  CONSTRAINT \"tbl_c\" UNIQUE (\"c\")\n" +
				")",
		},
		{
			name: "DEFAULT CURRENT_TIMESTAMP and doubled parens",
			input: "CREATE TABLE `t` (\n" +
				"  `a` int DEFAULT CURRENT_TIMESTAMP,\n" +
				"  `b` int NOT NULL DEFAULT ((7 + 11))\n" +
				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin",
			expected: "CREATE TABLE \"t\" (\n" +
				"  \"a\" integer DEFAULT (now()),\n" +
				"  \"b\" integer NOT NULL DEFAULT (7 + 11)\n" +
				")",
		},
		{
			name: "type-name substitutions",
			input: "CREATE TABLE `types` (\n" +
				"  `a` tinyint,\n" +
				"  `b` smallint,\n" +
				"  `c` mediumint,\n" +
				"  `d` bigint,\n" +
				"  `e` float,\n" +
				"  `f` double,\n" +
				"  `g` decimal(10,2),\n" +
				"  `h` blob,\n" +
				"  `i` datetime,\n" +
				"  `j` char(5)\n" +
				") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin",
			expected: "CREATE TABLE \"types\" (\n" +
				"  \"a\" smallint,\n" +
				"  \"b\" smallint,\n" +
				"  \"c\" integer,\n" +
				"  \"d\" bigint,\n" +
				"  \"e\" real,\n" +
				"  \"f\" double precision,\n" +
				"  \"g\" numeric,\n" +
				"  \"h\" bytea,\n" +
				"  \"i\" timestamp,\n" +
				"  \"j\" bpchar(5)\n" +
				")",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			actual := translateMysqlShowCreateTable(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestAutoQuotedIdentifiers verifies that identifiers needing quoting come
// out double-quoted in postgres-style
func TestAutoQuotedIdentifiers(t *testing.T) {
	type test struct {
		input    string
		expected []string
	}
	tests := []test{
		{
			input:    "SELECT * FROM `mydb/b2`.t",
			expected: []string{`select * from "mydb/b2".public.t`},
		},
		{
			input:    "INSERT INTO `mydb/b2`.t VALUES (1)",
			expected: []string{`insert into "mydb/b2".public.t values (1)`},
		},
		{
			input:    "USE `mydb/b2`",
			expected: []string{`USE "mydb/b2"`},
		},
		{
			input:    "SELECT `weird-col` FROM t",
			expected: []string{`SELECT "weird-col" FROM t`},
		},
	}

	for _, tc := range tests {
		t.Run(tc.input, func(t *testing.T) {
			actual := convertQuery(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestQualifiedNameConversion covers the MySQL→Postgres translation of
// `db.tbl` → `db.public.tbl` (and column-qualified equivalents).
func TestQualifiedNameConversion(t *testing.T) {
	type test struct {
		input    string
		expected []string
	}
	tests := []test{
		{
			input:    "CREATE TABLE test.x (pk int primary key)",
			expected: []string{"CREATE TABLE test.public.x (pk INTEGER NOT NULL PRIMARY KEY)"},
		},
		{
			input:    "INSERT INTO test.x VALUES (1, 2) on duplicate key update a = 5",
			expected: []string{"INSERT INTO test.public.x VALUES (1, 2) ON CONFLICT (fake) DO UPDATE SET a = 5"},
		},
		{
			input:    "SELECT pk FROM test.x",
			expected: []string{"select pk from test.public.x"},
		},
		{
			input:    "SELECT db1.t1.i FROM db1.t1 WHERE db1.t1.i > 0",
			expected: []string{"select db1.public.t1.i from db1.public.t1 where db1.public.t1.i > 0"},
		},
		{
			input:    "DELETE FROM test.x WHERE pk = 2",
			expected: []string{"delete from test.public.x where pk = 2"},
		},
		{
			input:    "UPDATE test.x SET pk = 300 WHERE pk = 3",
			expected: []string{"update test.public.x set pk = 300 where pk = 3"},
		},
		// Unqualified queries should pass through untouched by the qualified-name path.
		{
			input:    "CREATE TABLE x (pk int primary key)",
			expected: []string{"CREATE TABLE x (pk INTEGER NOT NULL PRIMARY KEY)"},
		},
	}

	for _, tc := range tests {
		t.Run(tc.input, func(t *testing.T) {
			actual := convertQuery(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestReplaceIntoConversion covers REPLACE INTO → INSERT ... ON CONFLICT
// DO UPDATE SET col = EXCLUDED.col translation.
func TestReplaceIntoConversion(t *testing.T) {
	type test struct {
		input    string
		expected []string
	}
	tests := []test{
		{
			input: "REPLACE INTO mytable (i, s) VALUES (1, 'first row')",
			expected: []string{
				"INSERT INTO mytable(i, s) VALUES (1, 'first row') ON CONFLICT (fake) DO UPDATE SET i = 1, s = 'first row'",
			},
		},
		{
			input: "REPLACE INTO mytable SET i = 1, s = 'first row'",
			expected: []string{
				"INSERT INTO mytable(i, s) VALUES (1, 'first row') ON CONFLICT (fake) DO UPDATE SET i = 1, s = 'first row'",
			},
		},
		{
			input: "REPLACE INTO mytable (s, i) VALUES ('x', 999)",
			expected: []string{
				"INSERT INTO mytable(s, i) VALUES ('x', 999) ON CONFLICT (fake) DO UPDATE SET s = 'x', i = 999",
			},
		},
	}

	for _, tc := range tests {
		t.Run(tc.input, func(t *testing.T) {
			actual := convertQuery(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestUpdateIgnoreConversion covers translating UPDATE IGNORE into an
// INSERT ... ON CONFLICT DO NOTHING form. Non-IGNORE UPDATEs are left alone.
func TestUpdateIgnoreConversion(t *testing.T) {
	type test struct {
		input    string
		expected []string
	}
	tests := []test{
		{
			input: "UPDATE IGNORE mytable SET i = 2 WHERE i = 1",
			expected: []string{
				"INSERT INTO mytable(i) SELECT 2 FROM mytable WHERE i = 1 ON CONFLICT (fake) DO NOTHING",
			},
		},
		{
			input: "UPDATE IGNORE pkTable SET pk = pk + 1, val = val + 1",
			expected: []string{
				`INSERT INTO "pkTable"(pk, val) SELECT pk + 1, val + 1 FROM "pkTable" ON CONFLICT (fake) DO NOTHING`,
			},
		},
		{
			input: "UPDATE IGNORE t1 SET v1 = v1 + 5 WHERE pk > 3",
			expected: []string{
				"INSERT INTO t1(v1) SELECT v1 + 5 FROM t1 WHERE pk > 3 ON CONFLICT (fake) DO NOTHING",
			},
		},
	}

	for _, tc := range tests {
		t.Run(tc.input, func(t *testing.T) {
			actual := convertQuery(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestAlterTableConversion covers the ALTER TABLE clauses translated by
// convertColumnAlter / convertConstraintAlter / convertIndexAlter.
func TestAlterTableConversion(t *testing.T) {
	type test struct {
		input    string
		expected []string
	}
	tests := []test{
		{
			input:    "ALTER TABLE foo ADD COLUMN c int",
			expected: []string{"ALTER TABLE foo ADD COLUMN c INTEGER NULL"},
		},
		{
			input:    "ALTER TABLE foo ADD COLUMN c int NOT NULL DEFAULT 7",
			expected: []string{"ALTER TABLE foo ADD COLUMN c INTEGER NOT NULL DEFAULT 7"},
		},
		{
			input:    "ALTER TABLE foo DROP COLUMN c",
			expected: []string{"ALTER TABLE foo DROP COLUMN c"},
		},
		{
			input:    "ALTER TABLE foo RENAME COLUMN c TO d",
			expected: []string{"ALTER TABLE foo RENAME COLUMN c TO d"},
		},
		{
			input: "ALTER TABLE foo MODIFY COLUMN c bigint NOT NULL",
			expected: []string{
				"ALTER TABLE foo ALTER COLUMN c SET DATA TYPE BIGINT",
				"ALTER TABLE foo ALTER COLUMN c SET NOT NULL",
			},
		},
		{
			input: "ALTER TABLE foo CHANGE COLUMN c d bigint",
			expected: []string{
				"ALTER TABLE foo ALTER COLUMN d SET DATA TYPE BIGINT",
				"ALTER TABLE foo ALTER COLUMN d DROP NOT NULL",
				"ALTER TABLE foo RENAME COLUMN c TO d",
			},
		},
		{
			input:    "ALTER TABLE foo RENAME TO bar",
			expected: []string{"ALTER TABLE foo RENAME TO bar"},
		},
		{
			input:    "ALTER TABLE foo DROP CONSTRAINT my_check",
			expected: []string{"ALTER TABLE foo DROP CONSTRAINT my_check"},
		},
		{
			input:    "ALTER TABLE foo DROP FOREIGN KEY my_fk",
			expected: []string{"ALTER TABLE foo DROP CONSTRAINT my_fk"},
		},
		{
			input: "ALTER TABLE foo ADD CONSTRAINT my_fk FOREIGN KEY (a) REFERENCES bar(b)",
			expected: []string{
				"ALTER TABLE foo ADD FOREIGN KEY (a) REFERENCES bar (b) ON DELETE RESTRICT ON UPDATE RESTRICT",
			},
		},
		{
			input:    "ALTER TABLE foo ADD INDEX my_idx (a, b)",
			expected: []string{"CREATE INDEX my_idx ON foo ( a ASC, b ASC ) NULLS NOT DISTINCT "},
		},
		{
			input:    "ALTER TABLE foo ADD UNIQUE my_uq (a)",
			expected: []string{"CREATE UNIQUE INDEX my_uq ON foo ( a ASC ) NULLS NOT DISTINCT "},
		},
		{
			input:    "ALTER TABLE foo DROP INDEX my_idx",
			expected: []string{"DROP INDEX foo@my_idx"},
		},
	}

	for _, tc := range tests {
		t.Run(tc.input, func(t *testing.T) {
			actual := convertQuery(tc.input)
			require.Equal(t, tc.expected, actual)
		})
	}
}

// TestExpressionConversionRobustness ensures convertExpr handles common
// MySQL expression kinds (AND/OR/NOT/IS NULL/IS NOT NULL/unary +-/qualified
// column references) without panicking. Previously any of these would crash
// the converter with "unhandled type" when the test framework tried to
// re-emit the parsed expression in postgres form.
func TestExpressionConversionRobustness(t *testing.T) {
	queries := []string{
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = -a",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = NOT b",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = b IS NULL",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = b IS NOT NULL",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = b AND b",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = b OR b",
		"INSERT INTO foo (a, b) VALUES (1, 2) ON DUPLICATE KEY UPDATE a = b DIV 2",
	}

	for _, q := range queries {
		t.Run(q, func(t *testing.T) {
			result := convertQuery(q)
			require.NotEmpty(t, result)
		})
	}
}

// TestBitTypeSupport tests BIT type conversion for dolt#9641 compatibility
// See: https://github.com/dolthub/dolt/issues/9641
func TestBitTypeSupport(t *testing.T) {
	result := convertQuery("CREATE TABLE bit_union_test_9641 (id INT PRIMARY KEY, flag BIT(1))")
	require.NotEmpty(t, result)
	require.Len(t, result, 1)
	require.Contains(t, result[0], "CREATE TABLE")
	require.Contains(t, result[0], "SMALLINT")

	bitResult := convertQuery("CREATE TABLE bit_test (id INT PRIMARY KEY, data BIT(8))")
	require.NotEmpty(t, bitResult)
	require.Contains(t, bitResult[0], "CREATE TABLE")

	insertResult := convertQuery("INSERT INTO bit_union_test_9641 VALUES (1, 0), (2, 1)")
	require.NotEmpty(t, insertResult)
	require.Len(t, insertResult, 1)
	require.Contains(t, insertResult[0], "INSERT INTO")

	unionResult := convertQuery("SELECT flag FROM bit_union_test_9641 WHERE id = 1 UNION SELECT NULL as flag")
	require.NotEmpty(t, unionResult)
	require.Len(t, unionResult, 1)
	require.Contains(t, unionResult[0], "UNION")
}

// AutoQuoteIdent takes an identifier and returns it quoted if necessary
func AutoQuoteIdent(name string) string {
	n := tree.Name(name)
	ctx := tree.NewFmtCtx(tree.FmtSimple)
	ctx.FormatNode(&n)
	return ctx.CloseAndGetString()
}