// Ensure that the input query is a Select statement that contains no Join,
// GroupBy, OrderBy, Limit or Distinct operations. Also ensure that the
// source table is present in the schema and has at least one primary key.
func (qs *QuerySplitter) validateQuery() error {
	statement, err := sqlparser.Parse(qs.query.Sql)
	if err != nil {
		return err
	}
	var ok bool
	qs.sel, ok = statement.(*sqlparser.Select)
	if !ok {
		return fmt.Errorf("not a select statement")
	}
	if qs.sel.Distinct != "" || qs.sel.GroupBy != nil ||
		qs.sel.Having != nil || len(qs.sel.From) != 1 ||
		qs.sel.OrderBy != nil || qs.sel.Limit != nil ||
		qs.sel.Lock != "" {
		return fmt.Errorf("unsupported query")
	}
	node, ok := qs.sel.From[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return fmt.Errorf("unsupported query")
	}
	qs.tableName = sqlparser.GetTableName(node.Expr)
	if qs.tableName == "" {
		return fmt.Errorf("not a simple table expression")
	}
	tableInfo, ok := qs.schemaInfo.tables[qs.tableName]
	if !ok {
		return fmt.Errorf("can't find table in schema")
	}
	if len(tableInfo.PKColumns) == 0 {
		return fmt.Errorf("no primary keys")
	}
	qs.pkCol = tableInfo.GetPKColumn(0).Name
	return nil
}

func buildUpdatePlan(upd *sqlparser.Update, schema *Schema) *Plan {
	plan := &Plan{
		ID:        NoPlan,
		Rewritten: generateQuery(upd),
	}
	tablename := sqlparser.GetTableName(upd.Table)
	plan.Table, plan.Reason = schema.LookupTable(tablename)
	if plan.Reason != "" {
		return plan
	}
	if plan.Table.Keyspace.ShardingScheme == Unsharded {
		plan.ID = UpdateUnsharded
		return plan
	}

	getWhereRouting(upd.Where, plan)
	switch plan.ID {
	case SelectSingleShardKey:
		plan.ID = UpdateSingleShardKey
	case SelectSingleLookup:
		plan.ID = UpdateSingleLookup
	case SelectMultiShardKey, SelectMultiLookup, SelectScatter:
		plan.ID = NoPlan
		plan.Reason = "too complex"
		return plan
	default:
		panic("unexpected")
	}
	if isIndexChanging(upd.Exprs, plan.Table.Indexes) {
		plan.ID = NoPlan
		plan.Reason = "index is changing"
	}
	return plan
}

func buildDeletePlan(del *sqlparser.Delete, schema *Schema) *Plan {
	plan := &Plan{
		ID:        NoPlan,
		Rewritten: generateQuery(del),
	}
	tablename := sqlparser.GetTableName(del.Table)
	plan.Table, plan.Reason = schema.FindTable(tablename)
	if plan.Reason != "" {
		return plan
	}
	if !plan.Table.Keyspace.Sharded {
		plan.ID = DeleteUnsharded
		return plan
	}

	getWhereRouting(del.Where, plan, true)
	switch plan.ID {
	case SelectEqual:
		plan.ID = DeleteEqual
		plan.Subquery = generateDeleteSubquery(del, plan.Table)
	case SelectIN, SelectScatter, SelectKeyrange:
		plan.ID = NoPlan
		plan.Reason = "delete has multi-shard where clause"
	default:
		panic("unexpected")
	}
	return plan
}

func buildDeletePlan(del *sqlparser.Delete, schema *Schema) *Plan {
	plan := &Plan{
		ID:        NoPlan,
		Rewritten: generateQuery(del),
	}
	tablename := sqlparser.GetTableName(del.Table)
	plan.Table, plan.Reason = schema.LookupTable(tablename)
	if plan.Reason != "" {
		return plan
	}
	if plan.Table.Keyspace.ShardingScheme == Unsharded {
		plan.ID = DeleteUnsharded
		return plan
	}

	getWhereRouting(del.Where, plan)
	switch plan.ID {
	case SelectSingleShardKey:
		plan.ID = DeleteSingleShardKey
	case SelectSingleLookup:
		plan.ID = DeleteSingleLookup
	case SelectMultiShardKey, SelectMultiLookup, SelectScatter:
		plan.ID = NoPlan
		plan.Reason = "too complex"
	default:
		panic("unexpected")
	}
	return plan
}

func buildUpdatePlan(upd *sqlparser.Update, schema *Schema) *Plan {
	plan := &Plan{
		ID:        NoPlan,
		Rewritten: generateQuery(upd),
	}
	tablename := sqlparser.GetTableName(upd.Table)
	plan.Table, plan.Reason = schema.FindTable(tablename)
	if plan.Reason != "" {
		return plan
	}
	if !plan.Table.Keyspace.Sharded {
		plan.ID = UpdateUnsharded
		return plan
	}

	getWhereRouting(upd.Where, plan, true)
	switch plan.ID {
	case SelectEqual:
		plan.ID = UpdateEqual
	case SelectIN, SelectScatter, SelectKeyrange:
		plan.ID = NoPlan
		plan.Reason = "update has multi-shard where clause"
		return plan
	default:
		panic("unexpected")
	}
	if isIndexChanging(upd.Exprs, plan.Table.ColVindexes) {
		plan.ID = NoPlan
		plan.Reason = "index is changing"
	}
	return plan
}

// newSplitParams validates and initializes all the fields except splitCount and
// numRowsPerQueryPart. It contains the common code for the constructors above.
func newSplitParams(sql string, bindVariables map[string]interface{}, splitColumns []string,
	schema map[string]*schema.Table) (*SplitParams, error) {

	statement, err := sqlparser.Parse(sql)
	if err != nil {
		return nil, fmt.Errorf("splitquery: failed parsing query: '%v', err: '%v'", sql, err)
	}
	selectAST, ok := statement.(*sqlparser.Select)
	if !ok {
		return nil, fmt.Errorf("splitquery: not a select statement")
	}
	if selectAST.Distinct != "" || selectAST.GroupBy != nil ||
		selectAST.Having != nil || len(selectAST.From) != 1 ||
		selectAST.OrderBy != nil || selectAST.Limit != nil ||
		selectAST.Lock != "" {
		return nil, fmt.Errorf("splitquery: unsupported query: %v", sql)
	}
	var aliasedTableExpr *sqlparser.AliasedTableExpr
	aliasedTableExpr, ok = selectAST.From[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return nil, fmt.Errorf("splitquery: unsupported FROM clause in query: %v", sql)
	}
	tableName := sqlparser.GetTableName(aliasedTableExpr.Expr)
	if tableName == "" {
		return nil, fmt.Errorf("splitquery: unsupported FROM clause in query"+
			" (must be a simple table expression): %v", sql)
	}
	tableSchema, ok := schema[tableName]
	if tableSchema == nil {
		return nil, fmt.Errorf("splitquery: can't find table in schema")
	}
	if len(splitColumns) == 0 {
		splitColumns = getPrimaryKeyColumns(tableSchema)
		if len(splitColumns) == 0 {
			panic(fmt.Sprintf("getPrimaryKeyColumns() returned an empty slice. %+v", tableSchema))
		}
	}
	if !areColumnsAPrefixOfAnIndex(splitColumns, tableSchema) {
		return nil, fmt.Errorf("splitquery: split-columns must be a prefix of the columns composing"+
			" an index. Sql: %v, split-columns: %v", sql, splitColumns)
	}
	// Get the split-columns types.
	splitColumnTypes := make([]querypb.Type, 0, len(splitColumns))
	for _, splitColumn := range splitColumns {
		i := tableSchema.FindColumn(splitColumn)
		if i == -1 {
			return nil, fmt.Errorf("can't find split-column: %v", splitColumn)
		}
		splitColumnTypes = append(splitColumnTypes, tableSchema.Columns[i].Type)
	}

	return &SplitParams{
		sql:              sql,
		bindVariables:    bindVariables,
		splitColumns:     splitColumns,
		splitColumnTypes: splitColumnTypes,
		selectAST:        selectAST,
		splitTableSchema: tableSchema,
	}, nil
}

func analyzeDelete(del *sqlparser.Delete, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanID:    PlanPassDML,
		FullQuery: GenerateFullQuery(del),
	}

	tableName := sqlparser.GetTableName(del.Table)
	if tableName == "" {
		plan.Reason = ReasonTable
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name.Lowered() != "primary" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = ReasonTableNoIndex
		return plan, nil
	}

	plan.OuterQuery = GenerateDeleteOuterQuery(del)

	if pkValues := analyzeWhere(del.Where, tableInfo.Indexes[0]); pkValues != nil {
		plan.PlanID = PlanDMLPK
		plan.PKValues = pkValues
		return plan, nil
	}

	plan.PlanID = PlanDMLSubquery
	plan.Subquery = GenerateDeleteSubquery(del, tableInfo)
	return plan, nil
}

func buildInsertPlan(ins *sqlparser.Insert, schema *Schema) *Plan {
	plan := &Plan{
		ID:        NoPlan,
		Rewritten: generateQuery(ins),
	}
	tablename := sqlparser.GetTableName(ins.Table)
	plan.Table, plan.Reason = schema.LookupTable(tablename)
	if plan.Reason != "" {
		return plan
	}
	if plan.Table.Keyspace.ShardingScheme == Unsharded {
		plan.ID = InsertUnsharded
		return plan
	}

	if len(ins.Columns) == 0 {
		plan.Reason = "no column list"
		return plan
	}
	var values sqlparser.Values
	switch rows := ins.Rows.(type) {
	case *sqlparser.Select, *sqlparser.Union:
		plan.Reason = "subqueries not allowed"
		return plan
	case sqlparser.Values:
		values = rows
	default:
		panic("unexpected")
	}
	if len(values) != 1 {
		plan.Reason = "multi-row inserts not supported"
		return plan
	}
	switch values[0].(type) {
	case *sqlparser.Subquery:
		plan.Reason = "subqueries not allowed"
		return plan
	}
	row := values[0].(sqlparser.ValTuple)
	if len(ins.Columns) != len(row) {
		plan.Reason = "column list doesn't match values"
		return plan
	}
	indexes := schema.Tables[tablename].Indexes
	plan.ID = InsertSharded
	plan.Values = make([]interface{}, 0, len(indexes))
	for _, index := range indexes {
		if err := buildIndexPlan(ins, tablename, index, plan); err != nil {
			plan.ID = NoPlan
			plan.Reason = err.Error()
			return plan
		}
	}
	// Query was rewritten
	plan.Rewritten = generateQuery(ins)
	return plan
}

func analyzeFrom(tableExprs sqlparser.TableExprs) (tablename string, hasHints bool) {
	if len(tableExprs) > 1 {
		return "", false
	}
	node, ok := tableExprs[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return "", false
	}
	return sqlparser.GetTableName(node.Expr), node.Hints != nil
}

func analyzeFrom(tableExprs sqlparser.TableExprs) string {
	if len(tableExprs) > 1 {
		return ""
	}
	node, ok := tableExprs[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return ""
	}
	return sqlparser.GetTableName(node.Expr)
}

func analyzeUpdate(upd *sqlparser.Update, getTable TableGetter) (plan *ExecPlan, err error) {
	// Default plan
	plan = &ExecPlan{
		PlanId:    PLAN_PASS_DML,
		FullQuery: GenerateFullQuery(upd),
	}

	tableName := sqlparser.GetTableName(upd.Table)
	if tableName == "" {
		plan.Reason = REASON_TABLE
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name != "PRIMARY" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = REASON_TABLE_NOINDEX
		return plan, nil
	}

	plan.SecondaryPKValues, err = analyzeUpdateExpressions(upd.Exprs, tableInfo.Indexes[0])
	if err != nil {
		if err == TooComplex {
			plan.Reason = REASON_PK_CHANGE
			return plan, nil
		}
		return nil, err
	}

	plan.PlanId = PLAN_DML_SUBQUERY
	plan.OuterQuery = GenerateUpdateOuterQuery(upd, tableInfo.Indexes[0])
	plan.Subquery = GenerateUpdateSubquery(upd, tableInfo)

	conditions := analyzeWhere(upd.Where)
	if conditions == nil {
		plan.Reason = REASON_WHERE
		return plan, nil
	}

	pkValues, err := getPKValues(conditions, tableInfo.Indexes[0])
	if err != nil {
		return nil, err
	}
	if pkValues != nil {
		plan.PlanId = PLAN_DML_PK
		plan.OuterQuery = plan.FullQuery
		plan.PKValues = pkValues
		return plan, nil
	}

	return plan, nil
}

// buildMinMaxQuery returns the query to execute to get the minimum and maximum of the splitColumn.
// The query returned is:
//   SELECT MIN(<splitColumn>), MAX(<splitColumn>) FROM <table>;
// where <table> is the table referenced in the original query (held in splitParams.sql).
func buildMinMaxQuery(splitParams *SplitParams) string {
	// The SplitParams constructor should have already checked that the FROM clause of the query
	// is a simple table expression, so this type-assertion should succeed.
	tableName := sqlparser.GetTableName(
		splitParams.selectAST.From[0].(*sqlparser.AliasedTableExpr).Expr)
	if tableName == "" {
		panic(fmt.Sprintf("Can't get tableName from query %v", splitParams.sql))
	}
	return fmt.Sprintf("select min(%v), max(%v) from %v",
		splitParams.splitColumns[0],
		splitParams.splitColumns[0],
		tableName)
}

func analyzeUpdate(upd *sqlparser.Update, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanID:    PlanPassDML,
		FullQuery: GenerateFullQuery(upd),
	}

	tableName := sqlparser.GetTableName(upd.Table)
	if tableName == "" {
		plan.Reason = ReasonTable
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name != "PRIMARY" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = ReasonTableNoIndex
		return plan, nil
	}

	plan.SecondaryPKValues, err = analyzeUpdateExpressions(upd.Exprs, tableInfo.Indexes[0])
	if err != nil {
		if err == ErrTooComplex {
			plan.Reason = ReasonPKChange
			return plan, nil
		}
		return nil, err
	}

	plan.OuterQuery = GenerateUpdateOuterQuery(upd)

	if conditions := analyzeWhere(upd.Where); conditions != nil {
		pkValues, err := getPKValues(conditions, tableInfo.Indexes[0])
		if err != nil {
			return nil, err
		}
		if pkValues != nil {
			plan.PlanID = PlanDMLPK
			plan.PKValues = pkValues
			return plan, nil
		}
	}

	plan.PlanID = PlanDMLSubquery
	plan.Subquery = GenerateUpdateSubquery(upd, tableInfo)
	return plan, nil
}

// Ensure that the input query is a Select statement that contains no Join,
// GroupBy, OrderBy, Limit or Distinct operations. Also ensure that the
// source table is present in the schema and has at least one primary key.
func (qs *QuerySplitter) validateQuery() error {
	statement, err := sqlparser.Parse(qs.sql)
	if err != nil {
		return err
	}
	var ok bool
	qs.sel, ok = statement.(*sqlparser.Select)
	if !ok {
		return fmt.Errorf("not a select statement")
	}
	if qs.sel.Distinct != "" || qs.sel.GroupBy != nil ||
		qs.sel.Having != nil || len(qs.sel.From) != 1 ||
		qs.sel.OrderBy != nil || qs.sel.Limit != nil ||
		qs.sel.Lock != "" {
		return fmt.Errorf("unsupported query")
	}
	node, ok := qs.sel.From[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return fmt.Errorf("unsupported query")
	}
	qs.tableName = sqlparser.GetTableName(node.Expr)
	if qs.tableName == "" {
		return fmt.Errorf("not a simple table expression")
	}
	tableInfo, ok := qs.schemaInfo.tables[qs.tableName]
	if !ok {
		return fmt.Errorf("can't find table in schema")
	}
	if len(tableInfo.PKColumns) == 0 {
		return fmt.Errorf("no primary keys")
	}
	if qs.splitColumn != "" {
		for _, index := range tableInfo.Indexes {
			for _, column := range index.Columns {
				if qs.splitColumn == column {
					return nil
				}
			}
		}
		return fmt.Errorf("split column is not indexed or does not exist in table schema, SplitColumn: %s, TableInfo.Table: %v", qs.splitColumn, tableInfo.Table)
	}
	qs.splitColumn = tableInfo.GetPKColumn(0).Name
	return nil
}

func analyzeDelete(del *sqlparser.Delete, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanId:    PLAN_PASS_DML,
		FullQuery: GenerateFullQuery(del),
	}

	tableName := sqlparser.GetTableName(del.Table)
	if tableName == "" {
		plan.Reason = REASON_TABLE
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name != "PRIMARY" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = REASON_TABLE_NOINDEX
		return plan, nil
	}

	plan.OuterQuery = GenerateDeleteOuterQuery(del)

	if conditions := analyzeWhere(del.Where); conditions != nil {
		pkValues, err := getPKValues(conditions, tableInfo.Indexes[0])
		if err != nil {
			return nil, err
		}
		if pkValues != nil {
			plan.PlanId = PLAN_DML_PK
			plan.PKValues = pkValues
			return plan, nil
		}
	}

	plan.PlanId = PLAN_DML_SUBQUERY
	plan.Subquery = GenerateDeleteSubquery(del, tableInfo)
	return plan, nil
}

func Fuzz(data []byte) int {
	stmt, err := sqlparser.Parse(string(data))
	if err != nil {
		if stmt != nil {
			panic("stmt is not nil on error")
		}
		return 0
	}
	if true {
		data1 := sqlparser.String(stmt)
		stmt1, err := sqlparser.Parse(data1)
		if err != nil {
			fmt.Printf("data0: %q\n", data)
			fmt.Printf("data1: %q\n", data1)
			panic(err)
		}
		if !fuzz.DeepEqual(stmt, stmt1) {
			fmt.Printf("data0: %q\n", data)
			fmt.Printf("data1: %q\n", data1)
			panic("not equal")
		}
	} else {
		sqlparser.String(stmt)
	}
	if sel, ok := stmt.(*sqlparser.Select); ok {
		var nodes []sqlparser.SQLNode
		for _, x := range sel.From {
			nodes = append(nodes, x)
		}
		for _, x := range sel.From {
			nodes = append(nodes, x)
		}
		for _, x := range sel.SelectExprs {
			nodes = append(nodes, x)
		}
		for _, x := range sel.GroupBy {
			nodes = append(nodes, x)
		}
		for _, x := range sel.OrderBy {
			nodes = append(nodes, x)
		}
		nodes = append(nodes, sel.Where)
		nodes = append(nodes, sel.Having)
		nodes = append(nodes, sel.Limit)
		for _, n := range nodes {
			if n == nil {
				continue
			}
			if x, ok := n.(sqlparser.SimpleTableExpr); ok {
				sqlparser.GetTableName(x)
			}
			if x, ok := n.(sqlparser.Expr); ok {
				sqlparser.GetColName(x)
			}
			if x, ok := n.(sqlparser.ValExpr); ok {
				sqlparser.IsValue(x)
			}
			if x, ok := n.(sqlparser.ValExpr); ok {
				sqlparser.IsColName(x)
			}
			if x, ok := n.(sqlparser.ValExpr); ok {
				sqlparser.IsSimpleTuple(x)
			}
			if x, ok := n.(sqlparser.ValExpr); ok {
				sqlparser.AsInterface(x)
			}
			if x, ok := n.(sqlparser.BoolExpr); ok {
				sqlparser.HasINClause([]sqlparser.BoolExpr{x})
			}
		}
	}
	buf := sqlparser.NewTrackedBuffer(nil)
	stmt.Format(buf)
	pq := buf.ParsedQuery()
	vars := map[string]interface{}{
		"A": 42,
		"B": 123123123,
		"C": "",
		"D": "a",
		"E": "foobar",
		"F": 1.1,
	}
	pq.GenerateQuery(vars)
	return 1
}

func analyzeInsert(ins *sqlparser.Insert, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanID:    PlanPassDML,
		FullQuery: GenerateFullQuery(ins),
	}
	tableName := sqlparser.GetTableName(ins.Table)
	if tableName == "" {
		plan.Reason = ReasonTable
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name.Lowered() != "primary" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = ReasonTableNoIndex
		return plan, nil
	}

	pkColumnNumbers := getInsertPKColumns(ins.Columns, tableInfo)

	if sel, ok := ins.Rows.(sqlparser.SelectStatement); ok {
		if ins.OnDup != nil {
			// Upserts not allowed for subqueries.
			// http://bugs.mysql.com/bug.php?id=58637
			plan.Reason = ReasonUpsert
			return plan, nil
		}
		plan.PlanID = PlanInsertSubquery
		plan.OuterQuery = GenerateInsertOuterQuery(ins)
		plan.Subquery = GenerateSelectLimitQuery(sel)
		if len(ins.Columns) != 0 {
			for _, col := range ins.Columns {
				colIndex := tableInfo.FindColumn(col.Original())
				if colIndex == -1 {
					return nil, fmt.Errorf("column %v not found in table %s", col, tableInfo.Name)
				}
				plan.ColumnNumbers = append(plan.ColumnNumbers, colIndex)
			}
		} else {
			// Add all columns.
			for colIndex := range tableInfo.Columns {
				plan.ColumnNumbers = append(plan.ColumnNumbers, colIndex)
			}
		}
		plan.SubqueryPKColumns = pkColumnNumbers
		return plan, nil
	}

	// If it's not a sqlparser.SelectStatement, it's Values.
	rowList := ins.Rows.(sqlparser.Values)
	pkValues, err := getInsertPKValues(pkColumnNumbers, rowList, tableInfo)
	if err != nil {
		return nil, err
	}
	if pkValues == nil {
		plan.Reason = ReasonComplexExpr
		return plan, nil
	}
	plan.PKValues = pkValues
	if ins.OnDup == nil {
		plan.PlanID = PlanInsertPK
		plan.OuterQuery = sqlparser.GenerateParsedQuery(ins)
		return plan, nil
	}
	if len(rowList) > 1 {
		// Upsert supported only for single row inserts.
		plan.Reason = ReasonUpsert
		return plan, nil
	}
	plan.SecondaryPKValues, err = analyzeUpdateExpressions(sqlparser.UpdateExprs(ins.OnDup), tableInfo.Indexes[0])
	if err != nil {
		plan.Reason = ReasonPKChange
		return plan, nil
	}
	plan.PlanID = PlanUpsertPK
	newins := *ins
	newins.Ignore = ""
	newins.OnDup = nil
	plan.OuterQuery = sqlparser.GenerateParsedQuery(&newins)
	upd := &sqlparser.Update{
		Comments: ins.Comments,
		Table:    ins.Table,
		Exprs:    sqlparser.UpdateExprs(ins.OnDup),
	}
	plan.UpsertQuery = GenerateUpdateOuterQuery(upd)
	return plan, nil
}

func analyzeInsert(ins *sqlparser.Insert, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanId:    PLAN_PASS_DML,
		FullQuery: GenerateFullQuery(ins),
	}
	tableName := sqlparser.GetTableName(ins.Table)
	if tableName == "" {
		plan.Reason = REASON_TABLE
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name != "PRIMARY" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = REASON_TABLE_NOINDEX
		return plan, nil
	}

	pkColumnNumbers := getInsertPKColumns(ins.Columns, tableInfo)

	if ins.OnDup != nil {
		// Upserts are not safe for statement based replication:
		// http://bugs.mysql.com/bug.php?id=58637
		plan.Reason = REASON_UPSERT
		return plan, nil
	}

	if sel, ok := ins.Rows.(sqlparser.SelectStatement); ok {
		plan.PlanId = PLAN_INSERT_SUBQUERY
		plan.OuterQuery = GenerateInsertOuterQuery(ins)
		plan.Subquery = GenerateSelectLimitQuery(sel)
		if len(ins.Columns) != 0 {
			plan.ColumnNumbers, err = analyzeSelectExprs(sqlparser.SelectExprs(ins.Columns), tableInfo)
			if err != nil {
				return nil, err
			}
		} else {
			// StarExpr node will expand into all columns
			n := sqlparser.SelectExprs{&sqlparser.StarExpr{}}
			plan.ColumnNumbers, err = analyzeSelectExprs(n, tableInfo)
			if err != nil {
				return nil, err
			}
		}
		plan.SubqueryPKColumns = pkColumnNumbers
		return plan, nil
	}

	// If it's not a sqlparser.SelectStatement, it's Values.
	rowList := ins.Rows.(sqlparser.Values)
	pkValues, err := getInsertPKValues(pkColumnNumbers, rowList, tableInfo)
	if err != nil {
		return nil, err
	}
	if pkValues != nil {
		plan.PlanId = PLAN_INSERT_PK
		plan.OuterQuery = plan.FullQuery
		plan.PKValues = pkValues
	}
	return plan, nil
}

func analyzeInsert(ins *sqlparser.Insert, getTable TableGetter) (plan *ExecPlan, err error) {
	plan = &ExecPlan{
		PlanID:    PlanPassDML,
		FullQuery: GenerateFullQuery(ins),
	}
	tableName := sqlparser.GetTableName(ins.Table)
	if tableName == "" {
		plan.Reason = ReasonTable
		return plan, nil
	}
	tableInfo, err := plan.setTableInfo(tableName, getTable)
	if err != nil {
		return nil, err
	}

	if len(tableInfo.Indexes) == 0 || tableInfo.Indexes[0].Name != "PRIMARY" {
		log.Warningf("no primary key for table %s", tableName)
		plan.Reason = ReasonTableNoIndex
		return plan, nil
	}

	pkColumnNumbers := getInsertPKColumns(ins.Columns, tableInfo)

	if sel, ok := ins.Rows.(sqlparser.SelectStatement); ok {
		if ins.OnDup != nil {
			// Upserts not allowed for subqueries.
			// http://bugs.mysql.com/bug.php?id=58637
			plan.Reason = ReasonUpsert
			return plan, nil
		}
		plan.PlanID = PlanInsertSubquery
		plan.OuterQuery = GenerateInsertOuterQuery(ins)
		plan.Subquery = GenerateSelectLimitQuery(sqlparser.ForUpdate(sel))
		if len(ins.Columns) != 0 {
			plan.ColumnNumbers, err = analyzeSelectExprs(sqlparser.SelectExprs(ins.Columns), tableInfo)
			if err != nil {
				return nil, err
			}
		} else {
			// StarExpr node will expand into all columns
			n := sqlparser.SelectExprs{&sqlparser.StarExpr{}}
			plan.ColumnNumbers, err = analyzeSelectExprs(n, tableInfo)
			if err != nil {
				return nil, err
			}
		}
		plan.SubqueryPKColumns = pkColumnNumbers
		return plan, nil
	}

	// If it's not a sqlparser.SelectStatement, it's Values.
	rowList := ins.Rows.(sqlparser.Values)
	pkValues, err := getInsertPKValues(pkColumnNumbers, rowList, tableInfo)
	if err != nil {
		return nil, err
	}
	if pkValues == nil {
		plan.Reason = ReasonComplexExpr
		return plan, nil
	}
	plan.PKValues = pkValues
	if ins.OnDup == nil {
		plan.PlanID = PlanInsertPK
		plan.OuterQuery = sqlparser.GenerateParsedQuery(ins)
		return plan, nil
	}
	if len(rowList) > 1 {
		// Upsert supported only for single row inserts.
		plan.Reason = ReasonUpsert
		return plan, nil
	}
	plan.SecondaryPKValues, err = analyzeUpdateExpressions(sqlparser.UpdateExprs(ins.OnDup), tableInfo.Indexes[0])
	if err != nil {
		if err == ErrTooComplex {
			plan.Reason = ReasonPKChange
			return plan, nil
		}
		return nil, err
	}
	plan.PlanID = PlanUpsertPK
	newins := *ins
	newins.Ignore = ""
	newins.OnDup = nil
	plan.OuterQuery = sqlparser.GenerateParsedQuery(&newins)
	upd := &sqlparser.Update{
		Comments: ins.Comments,
		Table:    ins.Table,
		Exprs:    sqlparser.UpdateExprs(ins.OnDup),
	}
	plan.UpsertQuery = GenerateUpdateOuterQuery(upd)
	return plan, nil
}

// newSplitParams validates and initializes all the fields except splitCount and
// numRowsPerQueryPart. It contains the common code for the constructors above.
func newSplitParams(
	sql string,
	bindVariables map[string]interface{},
	splitColumnNames []sqlparser.ColIdent,
	schemaMap map[string]*schema.Table,
) (*SplitParams, error) {
	statement, err := sqlparser.Parse(sql)
	if err != nil {
		return nil, fmt.Errorf("failed parsing query: '%v', err: '%v'", sql, err)
	}
	selectAST, ok := statement.(*sqlparser.Select)
	if !ok {
		return nil, fmt.Errorf("not a select statement")
	}
	if selectAST.Distinct != "" || selectAST.GroupBy != nil ||
		selectAST.Having != nil || len(selectAST.From) != 1 ||
		selectAST.OrderBy != nil || selectAST.Limit != nil ||
		selectAST.Lock != "" {
		return nil, fmt.Errorf("unsupported query: %v", sql)
	}
	var aliasedTableExpr *sqlparser.AliasedTableExpr
	aliasedTableExpr, ok = selectAST.From[0].(*sqlparser.AliasedTableExpr)
	if !ok {
		return nil, fmt.Errorf("unsupported FROM clause in query: %v", sql)
	}
	tableName := sqlparser.GetTableName(aliasedTableExpr.Expr)
	if tableName == "" {
		return nil, fmt.Errorf("unsupported FROM clause in query"+
			" (must be a simple table expression): %v", sql)
	}
	tableSchema, ok := schemaMap[tableName]
	if tableSchema == nil {
		return nil, fmt.Errorf("can't find table in schema")
	}

	// Get the schema.TableColumn representation of each splitColumnName.
	var splitColumns []*schema.TableColumn
	if len(splitColumnNames) == 0 {
		splitColumns = getPrimaryKeyColumns(tableSchema)
	} else {
		splitColumns, err = findSplitColumnsInSchema(splitColumnNames, tableSchema)
		if err != nil {
			return nil, err
		}
		if !areColumnsAPrefixOfAnIndex(splitColumns, tableSchema) {
			return nil, fmt.Errorf("split-columns must be a prefix of the columns composing"+
				" an index. Sql: %v, split-columns: %v", sql, splitColumns)
		}
	}

	if len(splitColumns) == 0 {
		panic(fmt.Sprintf(
			"Empty set of split columns. splitColumns: %+v, tableSchema: %+v",
			splitColumns, tableSchema))
	}

	return &SplitParams{
		sql:              sql,
		bindVariables:    bindVariables,
		splitColumns:     splitColumns,
		selectAST:        selectAST,
		splitTableSchema: tableSchema,
	}, nil
}