func (sq *SqlQuery) allowQueries(dbconfig *eproto.DBConfigs) {
	sq.statemu.Lock()
	v := sq.state.Get()
	switch v {
	case ABORT, SERVING:
		sq.statemu.Unlock()
		log.Info("Ignoring allowQueries request, current state: %v", v)
		return
	case INITIALIZING, SHUTTING_DOWN:
		panic("unreachable")
	}
	// state is NOT_SERVING
	sq.setState(INITIALIZING)

	defer func() {
		if x := recover(); x != nil {
			log.Error("%s", x.(*TabletError).Message)
			sq.setState(NOT_SERVING)
			return
		}
		sq.setState(SERVING)
	}()

	sq.qe.Open(dbconfig)
	sq.dbconfig = dbconfig
	sq.sessionId = Rand()
	log.Info("Session id: %d", sq.sessionId)
}

func (em *Engine) Init(conf *proto.DBConfigs) error {
	log.Info("Begin init engine")
	if err := em.dbEngine.Init(conf); err != nil {
		log.Info("Init engine %v error, %v", em.dbEngine.Name(), err)
		return proto.ErrDbInitError
	}
	log.Info("Init engine %v complete", em.dbEngine.Name())
	return nil
}

//-----------------------------------------------
// Execution
func (qe *QueryEngine) execInsertPK(logStats *sqlQueryStats, conn PoolConnection, plan *CompiledPlan) (qr *eproto.QueryResult) {
	log.Info("Execute insert pk sql %s", plan.Query)
	tableName := plan.TableName
	tableInfo := qe.schemaInfo.tables[plan.TableName]
	rowColumns := plan.RowColumns
	var key []byte
	var columnName string
	keys := make([][]byte, 0, len(rowColumns)*len(tableInfo.Columns))
	values := make([][]byte, 0, len(rowColumns)*len(tableInfo.Columns))
	pkList := buildValueList(tableInfo, plan.PKValues, plan.BindVars)
	for i, columnsMap := range rowColumns {
		pkvalue := buildPkValue(pkList[i])
		log.Info("Pk Value is %v", string(pkvalue))
		for _, columnDef := range tableInfo.Columns {
			columnName = columnDef.Name
			if columnDef.IsPk {
				key = buildTableRowColumnKey(tableName, columnName, pkvalue)
				log.Info("pk key is %v", string(key))
				keys = append(keys, key)
				values = append(values, []byte{'0'})
			} else if columnDef.IsAuto {
				if _, ok := columnsMap[columnName]; ok {
					panic(NewTabletErrorDB(FAIL, fmt.Errorf("field %s value is auto created", columnName)))
				}
				// TODO
			} else {
				value, ok := columnsMap[columnName]
				if !ok {
					if !columnDef.Nullable {
						panic(NewTabletErrorDB(FAIL, fmt.Errorf("column %s shouldn't be null", columnDef.Name)))
					}
				}
				if !value.(sqltypes.Value).IsNull() {
					key = buildTableRowColumnKey(tableName, columnName, pkvalue)
					log.Info("normal key is %v", string(key))
					keys = append(keys, key)
					values = append(values, value.(sqltypes.Value).Raw())
					log.Info("normal value is %v", value.(sqltypes.Value).String())
				}
			}
		}
	}
	atomic.AddInt64(&qe.activeConnection, 1)
	defer atomic.AddInt64(&qe.activeConnection, -1)
	err := conn.Puts(nil, keys, values)
	if err != nil {
		panic(NewTabletErrorDB(FAIL, err))
	}

	qr = &eproto.QueryResult{RowsAffected: uint64(len(rowColumns))}
	return qr
}

func (si *SchemaInfo) GetPlan(logStats *sqlQueryStats, sql string) (plan *ExecPlan) {
	log.Warn("plan sql %v", sql)
	si.mu.Lock()
	defer si.mu.Unlock()
	if plan := si.getQuery(sql); plan != nil {
		return plan
	}

	var tableInfo *schema.Table
	GetTable := func(tableName string) (table *schema.Table, ok bool) {
		tableInfo, ok = si.tables[tableName]
		if !ok {
			return nil, false
		}
		return tableInfo, true
	}
	splan, err := sqlparser.ExecParse(sql, GetTable)
	if err != nil {
		log.Info("parse error %v", err.Error())
		panic(NewTabletError(FAIL, "%s", err))
	}
	plan = &ExecPlan{ExecPlan: splan, Table: tableInfo}
	if plan.PlanId.IsSelect() {
		fields := make([]eproto.Field, len(plan.ColumnNumbers))
		for i, cIdx := range plan.ColumnNumbers {
			column := si.tables[plan.TableName].Columns[cIdx]
			fields[i] = eproto.Field{column.Name, int64(column.Category)}
		}
		plan.Fields = fields
	} else if plan.PlanId == sqlparser.PLAN_DDL || plan.PlanId == sqlparser.PLAN_SET {
		return plan
	}
	si.queries.Set(sql, plan)
	return plan
}

func (si *SchemaInfo) DropTable(tableName string) {
	si.mu.Lock()
	defer si.mu.Unlock()
	delete(si.tables, tableName)
	si.queries.Clear()
	log.Info("Table %s forgotten", tableName)
}

func (qe *QueryEngine) selectAll(logStats *sqlQueryStats, plan *CompiledPlan) (result *eproto.QueryResult) {
	tableName := plan.TableName
	pks := qe.getAllPks(logStats, plan.TableName, nil)
	var keys, values [][]byte
	for _, pk := range pks {
		for _, field := range plan.Fields {
			keys = append(keys, buildTableRowColumnKey(tableName, field.Name, pk))
		}
	}

	result = &eproto.QueryResult{}
	result.Fields = plan.Fields

	if len(pks) == 0 {
		result.RowsAffected = 0
		return result
	}

	values = qe.fetch(logStats, keys)

	rowList := make([][]sqltypes.Value, len(pks))
	for i := range pks {
		rowList[i] = make([]sqltypes.Value, len(plan.Fields))
		for j, field := range plan.Fields {
			rowList[i][j] = buildValue(values[i*len(plan.Fields)+j], field.Type)
			log.Info(rowList[i][j].String())
		}
	}
	result.Rows = rowList
	result.RowsAffected = uint64(len(pks))
	return result
}

func (node *Node) execAnalyzeValue() *Node {
	log.Info("node.Type %v", node.Type)
	switch node.Type {
	case STRING, NUMBER, VALUE_ARG:
		return node
	}
	return nil
}

func ExecParse(sql string, getTable TableGetter) (plan *ExecPlan, err error) {
	defer handleError(&err)

	tree, err := Parse(sql)

	log.Info("tree: %v", tree.Type)
	log.Info("%v", tree.TreeString())

	if err != nil {
		return nil, err
	}
	plan = tree.execAnalyzeSql(getTable)
	if plan.PlanId == PLAN_PASS_DML {
		log.Warn("PASS_DML: %s", sql)
	}
	return plan, nil
}

func (qe *QueryEngine) Open(config *eproto.DBConfigs) {
	// Wait for Close, in case it's running
	qe.mu.Lock()
	defer qe.mu.Unlock()
	err := qe.engine.Init(config)
	if err != nil {
		log.Info(err.Error())
		panic(NewTabletErrorDB(FATAL, err))
	}
	connFactory := ConnectionCreator(config.AppConnectParams, qe.engine)
	qe.connPool.Open(connFactory)
	// qe.streamConnPool.Open(connFactory)
	// qe.reservedPool.Open(connFactory)
	start := time.Now().UnixNano()
	qe.schemaInfo.Open(connFactory)
	log.Info("Time taken to load the schema: %v ms", (time.Now().UnixNano()-start)/1e6)
}

// No condition in the sql
// if modiftyType is true, update method, else delete method
func (qe *QueryEngine) execModifyAll(logStats *sqlQueryStats, conn PoolConnection, plan *CompiledPlan, modifyType bool) *eproto.QueryResult {
	pks := qe.getAllPks(logStats, plan.TableName, conn)
	log.Info("delete %d", len(pks))
	if modifyType {
		return qe.execUpdate(logStats, conn, plan, pks)
	} else {
		return qe.execDelete(logStats, conn, plan, pks)
	}
}

func (qe *QueryEngine) getAllPks(logStats *sqlQueryStats, tableName string, conn PoolConnection) (pks [][]byte) {
	tableInfo := qe.schemaInfo.tables[tableName]
	pkStart := []byte(fmt.Sprintf("%s|%s|", tableName, tableInfo.GetPk().Name))
	pkEnd := []byte(fmt.Sprintf("%s|%s||", tableName, tableInfo.GetPk().Name))
	_, pks = qe.fetchIterate(logStats, conn, pkStart, pkEnd, 0, true, false)
	for _, pk := range pks {
		log.Info("pkpkpk:%s", string(pk))
	}
	return pks
}

func (sq *SqlQuery) Execute(context *rpcproto.Context, query *proto.Query, reply *eproto.QueryResult) (err error) {
	log.Info("sql is %v", query.Sql)
	logStats := newSqlQueryStats("Execute", context)
	defer handleExecError(query, &err, logStats)

	sq.checkState(query.SessionId)

	*reply = *sq.qe.Execute(logStats, query)
	return nil
}

func (em *Engine) Connect(params *proto.DbConnectParams) (conn *DBConnection, err error) {
	var c proto.DbConnection
	c, err = em.dbEngine.Connect(params)
	log.Info("Come to a new client %v, id is %d", params.UserName, c.Id())
	if conn != nil && err != nil {
		conn.Close()
		return nil, err
	}
	conn = &DBConnection{connectionParams: params, DbConnection: c}
	return conn, nil
}

// Load loads the contents of a JSON file named
// filename into c.
func (c *cramMD5Credentials) Load(filename string) error {
	data, err := ioutil.ReadFile(filename)
	if err != nil {
		return err
	}
	if err = json.Unmarshal(data, c); err != nil {
		return err
	}
	log.Info("Loaded credentials from %s.", filename)
	return nil
}

func NewEngine(name string) (*Engine, error) {
	em := new(Engine)
	if engineInit, ok := engineImpls[name]; ok {
		log.Info("Get Engine : %v", name)
		engine := engineInit()
		em.dbEngine = engine
		return em, nil
	} else {
		return nil, proto.ErrUnknownDBEngineName
	}
}

func main() {
	flag.Parse()
	servenv.Init()

	ts.InitQueryService()

	dbConfigs := &proto.DBConfigs{DataPath: "testrock"}
	dbConfigs.AppConnectParams = &proto.DbConnectParams{DbName: "test", UserName: "li"}
	ts.AllowQueries(dbConfigs)

	log.Info("starting vtocc %v", *port)
	servenv.OnClose(func() {
		time.Sleep(5 * time.Millisecond)
		ts.DisallowQueries()
	})
	servenv.Run(*port)
}

// RunSecure is like Run, but it additionally listens for RPC and HTTP
// requests using TLS on securePort, using the passed certificate,
// key, and CA certificate.
func RunSecure(port int, securePort int, cert, key, caCert string) {
	onRunHooks.Fire()
	ServeRPC()

	l, err := proc.Listen(fmt.Sprintf("%v", port))
	if err != nil {
		log.Critical(err.Error())
	}

	go http.Serve(l, nil)

	if securePort != 0 {
		log.Info("listening on secure port %v", securePort)
		SecureServe(fmt.Sprintf(":%d", securePort), cert, key, caCert)
	}
	proc.Wait()
	Close()
}

func (node *Node) execAnalyzeUpdateExpressions(pkIndex *schema.Index) (updateColumns []Pair, status updateStatus) {
	updateColumns = make([]Pair, node.Len())
	for i := 0; i < node.Len(); i++ {
		columnName := string(node.At(i).At(0).Value)
		index := pkIndex.FindColumn(columnName)
		if index != -1 {
			// update pkvalues not supported
			return nil, updatePkValue
		}
		value := node.At(i).At(1).execAnalyzeValue()
		if value == nil {
			log.Warn("unsupported update expression", node.At(i).At(0))
			return nil, updateUnsupport
		}
		log.Info(string(value.Value))
		updateColumns[i] = Pair{columnName, string(value.Value)}
	}
	return updateColumns, updateOk
}

func (sq *SqlQuery) disallowQueries() {
	sq.statemu.Lock()
	defer sq.statemu.Unlock()
	switch sq.state.Get() {
	case NOT_SERVING, ABORT:
		return
	case INITIALIZING, SHUTTING_DOWN:
		panic("unreachable")
	}
	// state is SERVING
	sq.setState(SHUTTING_DOWN)
	defer func() {
		sq.setState(NOT_SERVING)
	}()

	log.Info("Stopping query service: %d", sq.sessionId)
	sq.qe.Close()
	sq.sessionId = 0
	sq.dbconfig = nil
}

func (qe *QueryEngine) selectPkEqual(logStats *sqlQueryStats, plan *CompiledPlan) (result *eproto.QueryResult) {
	pkRows := buildPkValueList(plan.Table, plan.PKValues, plan.BindVars)
	if len(pkRows) != 1 || plan.Fields == nil {
		panic("unexpected")
	}

	tableName := plan.TableName
	keys := make([][]byte, len(plan.Fields))
	pkValue := buildCompositeValue(pkRows[0])
	log.Info("pk value is %s", string(pkValue))

	result = &eproto.QueryResult{}
	result.Fields = plan.Fields

	// check if pk exists
	if !qe.fetchExists(logStats, buildTableRowColumnKey(tableName, plan.Table.GetPk().Name, pkValue)) {
		// not exists
		result.RowsAffected = 0
		return result
	}

	for i, field := range plan.Fields {
		keys[i] = buildTableRowColumnKey(tableName, field.Name, pkValue)
	}

	values := qe.fetch(logStats, keys)

	rowValues := make([]sqltypes.Value, len(plan.Fields))
	for i, field := range plan.Fields {
		if len(values[i]) != 0 {
			rowValues[i] = buildValue(values[i], field.Type)
		} else {
			rowValues[i] = sqltypes.NULL
		}
	}

	result.Rows = make([][]sqltypes.Value, 1)
	result.Rows[0] = rowValues
	result.RowsAffected = 1
	return result
}