// exec executes the request. Any error encountered is returned; it is
// the caller's responsibility to update the response.
func (e *Executor) execStmts(sql string, planMaker *planner) driver.Response {
	var resp driver.Response
	stmts, err := planMaker.parser.Parse(sql, parser.Syntax(planMaker.session.Syntax))
	if err != nil {
		// A parse error occurred: we can't determine if there were multiple
		// statements or only one, so just pretend there was one.
		resp.Results = append(resp.Results, makeResultFromError(planMaker, err))
		return resp
	}
	for _, stmt := range stmts {
		result, err := e.execStmt(stmt, planMaker)
		if err != nil {
			result = makeResultFromError(planMaker, err)
		}
		resp.Results = append(resp.Results, result)
		// Release the leases once a transaction is complete.
		if planMaker.txn == nil {
			planMaker.releaseLeases(e.db)

			// The previous transaction finished executing some schema changes. Wait for
			// the schema changes to propagate to all nodes, so that once the executor
			// returns the new schema are live everywhere. This is not needed for
			// correctness but is done to make the UI experience/tests predictable.
			if err := e.waitForCompletedSchemaChangesToPropagate(planMaker); err != nil {
				log.Warning(err)
			}
		}
	}
	return resp
}

// exec executes the request. Any error encountered is returned; it is
// the caller's responsibility to update the response.
func (s *Server) exec(req driver.Request) (driver.Response, error) {
	var resp driver.Response

	// Pick up current session state.
	planner := planner{db: s.db}
	if req.Session != nil {
		// TODO(tschottdorf) will have to validate the Session information (for
		// instance, whether access to the stored database is permitted).
		if err := gogoproto.Unmarshal(req.Session, &planner.session); err != nil {
			return resp, err
		}
	}
	stmts, err := parser.Parse(req.Sql)
	if err != nil {
		return resp, err
	}
	for _, stmt := range stmts {
		// Bind all the placeholder variables in the stmt to actual values.
		if err := parser.FillArgs(stmt, parameters(req.Params)); err != nil {
			return resp, err
		}
		var plan planNode
		if plan, err = planner.makePlan(stmt); err != nil {
			return resp, err
		}

		result := driver.Result{
			Columns: plan.Columns(),
		}
		for plan.Next() {
			values := plan.Values()
			row := driver.Result_Row{}
			row.Values = make([]driver.Datum, 0, len(values))
			for _, val := range values {
				switch vt := val.(type) {
				case parser.DBool:
					row.Values = append(row.Values, driver.Datum{BoolVal: (*bool)(&vt)})
				case parser.DInt:
					row.Values = append(row.Values, driver.Datum{IntVal: (*int64)(&vt)})
				case parser.DFloat:
					row.Values = append(row.Values, driver.Datum{FloatVal: (*float64)(&vt)})
				case parser.DString:
					row.Values = append(row.Values, driver.Datum{StringVal: (*string)(&vt)})
				case parser.DNull:
					row.Values = append(row.Values, driver.Datum{})
				default:
					return resp, util.Errorf("unsupported datum: %T", val)
				}
			}
			result.Rows = append(result.Rows, row)
		}
		resp.Results = append(resp.Results, result)
	}

	// Update session state.
	resp.Session, err = gogoproto.Marshal(&planner.session)
	return resp, err
}

// exec executes the request. Any error encountered is returned; it is
// the caller's responsibility to update the response.
func (e *Executor) execStmts(sql string, planMaker *planner) driver.Response {
	var resp driver.Response
	stmts, err := planMaker.parser.Parse(sql, parser.Syntax(planMaker.session.Syntax))
	if err != nil {
		// A parse error occurred: we can't determine if there were multiple
		// statements or only one, so just pretend there was one.
		resp.Results = append(resp.Results, makeResultFromError(planMaker, err))
		return resp
	}
	for _, stmt := range stmts {
		result, err := e.execStmt(stmt, planMaker)
		if err != nil {
			result = makeResultFromError(planMaker, err)
		}
		// Release the leases once a transaction is complete.
		if planMaker.txn == nil {
			planMaker.releaseLeases(e.db)
			// Execute any schema changes that were scheduled.
			if len(planMaker.schemaChangers) > 0 &&
				// Disable execution in some tests.
				!disableSyncSchemaChangeExec {
				retryOpts := retry.Options{
					InitialBackoff: 20 * time.Millisecond,
					MaxBackoff:     200 * time.Millisecond,
					Multiplier:     2,
				}
				for _, sc := range planMaker.schemaChangers {
					sc.db = e.db
					for r := retry.Start(retryOpts); r.Next(); {
						if done, err := sc.IsDone(); err != nil {
							log.Warning(err)
							break
						} else if done {
							break
						}
						if err := sc.exec(); err != nil {
							if err == errExistingSchemaChangeLease {
								// Try again.
								continue
							}
							// All other errors can be reported.
							result = makeResultFromError(planMaker, err)
						}
						break
					}
				}
			}
		}
		resp.Results = append(resp.Results, result)
	}
	return resp
}

// ShowTables returns all the tables.
func (s *Server) ShowTables(session *Session, p *parser.ShowTables, args []driver.Datum, resp *driver.Response) error {
	if p.Name == nil {
		if session.Database == "" {
			return errNoDatabase
		}
		p.Name = append(p.Name, session.Database)
	}
	dbID, err := s.lookupDatabase(p.Name.String())
	if err != nil {
		return err
	}
	prefix := keys.MakeNameMetadataKey(dbID, "")
	sr, err := s.db.Scan(prefix, prefix.PrefixEnd(), 0)
	if err != nil {
		return err
	}
	var rows []driver.Result_Row

	for _, row := range sr {
		name := string(bytes.TrimPrefix(row.Key, prefix))
		rows = append(rows, driver.Result_Row{
			Values: []driver.Datum{
				{StringVal: &name},
			},
		})
	}
	resp.Results = []driver.Result{
		{
			Columns: []string{"tables"},
			Rows:    rows,
		},
	}
	return nil
}

// ShowIndex returns all the indexes for a table.
func (s *Server) ShowIndex(session *Session, p *parser.ShowIndex, args []driver.Datum, resp *driver.Response) error {
	desc, err := s.getTableDesc(session.Database, p.Table)
	if err != nil {
		return err
	}

	// TODO(pmattis): This output doesn't match up with MySQL. Should it?
	var rows []driver.Result_Row

	name := p.Table.Table()
	for i, index := range desc.Indexes {
		for j, col := range index.ColumnNames {
			seq := int64(j + 1)
			c := col
			rows = append(rows, driver.Result_Row{
				Values: []driver.Datum{
					{StringVal: &name},
					{StringVal: &desc.Indexes[i].Name},
					{BoolVal: &desc.Indexes[i].Unique},
					{IntVal: &seq},
					{StringVal: &c},
				},
			})
		}
	}
	resp.Results = []driver.Result{
		{
			// TODO(pmattis): This output doesn't match up with MySQL. Should it?
			Columns: []string{"Table", "Name", "Unique", "Seq", "Column"},
			Rows:    rows,
		},
	}
	return nil
}

// ShowColumns of a table
func (s *Server) ShowColumns(session *Session, p *parser.ShowColumns, args []driver.Datum, resp *driver.Response) error {
	desc, err := s.getTableDesc(session.Database, p.Table)
	if err != nil {
		return err
	}
	var rows []driver.Result_Row
	for i, col := range desc.Columns {
		t := col.Type.SQLString()
		rows = append(rows, driver.Result_Row{
			Values: []driver.Datum{
				{StringVal: &desc.Columns[i].Name},
				{StringVal: &t},
				{BoolVal: &desc.Columns[i].Nullable},
			},
		})
	}
	// TODO(pmattis): This output doesn't match up with MySQL. Should it?
	resp.Results = []driver.Result{
		{
			Columns: []string{"Field", "Type", "Null"},
			Rows:    rows,
		},
	}
	return nil
}

// exec executes the request. Any error encountered is returned; it is
// the caller's responsibility to update the response.
func (s server) execStmts(sql string, params parameters, planMaker *planner) driver.Response {
	var resp driver.Response
	stmts, err := parser.Parse(sql, parser.Syntax(planMaker.session.Syntax))
	if err != nil {
		// A parse error occured: we can't determine if there were multiple
		// statements or only one, so just pretend there was one.
		resp.Results = append(resp.Results, rollbackTxnAndReturnResultWithError(planMaker, err))
		return resp
	}
	for _, stmt := range stmts {
		result, err := s.execStmt(stmt, params, planMaker)
		if err != nil {
			result = rollbackTxnAndReturnResultWithError(planMaker, err)
		}
		resp.Results = append(resp.Results, result)
	}
	return resp
}

// exec executes the request. Any error encountered is returned; it is
// the caller's responsibility to update the response.
func (e *Executor) execStmts(sql string, params parameters, planMaker *planner) driver.Response {
	var resp driver.Response
	stmts, err := parser.Parse(sql, parser.Syntax(planMaker.session.Syntax))
	if err != nil {
		// A parse error occurred: we can't determine if there were multiple
		// statements or only one, so just pretend there was one.
		resp.Results = append(resp.Results, makeResultFromError(planMaker, err))
		return resp
	}
	for _, stmt := range stmts {
		result, err := e.execStmt(stmt, params, planMaker)
		if err != nil {
			result = makeResultFromError(planMaker, err)
		}
		resp.Results = append(resp.Results, result)
		// TODO(pmattis): Is this the correct time to be releasing leases acquired
		// during execution of the statement?
		//
		// TODO(pmattis): Need to record the leases used by a transaction within
		// the transaction state and restore it when the transaction is restored.
		planMaker.releaseLeases(e.db)
	}
	return resp
}

// ShowDatabases returns all the databases.
func (s *Server) ShowDatabases(session *Session, p *parser.ShowDatabases, args []driver.Datum, resp *driver.Response) error {
	prefix := keys.MakeNameMetadataKey(structured.RootNamespaceID, "")
	sr, err := s.db.Scan(prefix, prefix.PrefixEnd(), 0)
	if err != nil {
		return err
	}
	var rows []driver.Result_Row
	for _, row := range sr {
		name := string(bytes.TrimPrefix(row.Key, prefix))
		rows = append(rows, driver.Result_Row{
			Values: []driver.Datum{
				{StringVal: &name},
			},
		})
	}
	resp.Results = []driver.Result{
		{
			Columns: []string{"Database"},
			Rows:    rows,
		},
	}
	return nil
}

// Select selects rows from a single table.
func (s *Server) Select(session *Session, p *parser.Select, args []driver.Datum, resp *driver.Response) error {
	if len(p.Exprs) != 1 {
		return fmt.Errorf("TODO(pmattis): unsupported select exprs: %s", p.Exprs)
	}
	if _, ok := p.Exprs[0].(*parser.StarExpr); !ok {
		return fmt.Errorf("TODO(pmattis): unsupported select expr: %s", p.Exprs)
	}

	if len(p.From) != 1 {
		return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
	}
	var desc *structured.TableDescriptor
	{
		ate, ok := p.From[0].(*parser.AliasedTableExpr)
		if !ok {
			return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		table, ok := ate.Expr.(parser.QualifiedName)
		if !ok {
			return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		var err error
		desc, err = s.getTableDesc(session.Database, table)
		if err != nil {
			return err
		}
	}

	// Retrieve all of the keys that start with our index key prefix.
	startKey := proto.Key(encodeIndexKeyPrefix(desc.ID, desc.Indexes[0].ID))
	endKey := startKey.PrefixEnd()
	sr, err := s.db.Scan(startKey, endKey, 0)
	if err != nil {
		return err
	}

	// All of the columns for a particular row will be grouped together. We loop
	// over the returned key/value pairs and decode the key to extract the
	// columns encoded within the key and the column ID. We use the column ID to
	// lookup the column and decode the value. All of these values go into a map
	// keyed by column name. When the index key changes we output a row
	// containing the current values.
	//
	// The TODOs here are too numerous to list. This is only performing a full
	// table scan using the primary key.

	var rows []driver.Result_Row
	var primaryKey []byte
	vals := valMap{}
	l := len(sr)
	// Iterate through the scan result set. We decide at the beginning of each
	// new row whether the previous row is to be output. To deal with the very
	// last one, the loop below goes an extra iteration (i==l).
	for i := 0; ; i++ {
		var kv client.KeyValue
		if i < l {
			kv = sr[i]
		}
		if primaryKey != nil && (i == l || !bytes.HasPrefix(kv.Key, primaryKey)) {
			// The current key belongs to a new row. Decide whether the last
			// row is to be output.
			if output, err := shouldOutputRow(p.Where, vals); err != nil {
				return err
			} else if output {
				rows = append(rows, outputRow(desc.Columns, vals))
			}
			vals = valMap{}
		}

		if i >= l {
			break
		}

		remaining, err := decodeIndexKey(desc, desc.Indexes[0], vals, kv.Key)
		if err != nil {
			return err
		}
		primaryKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		_, colID := encoding.DecodeUvarint(remaining)
		if err != nil {
			return err
		}
		col, err := desc.FindColumnByID(uint32(colID))
		if err != nil {
			return err
		}
		vals[col.Name] = unmarshalValue(*col, kv)

		if log.V(2) {
			log.Infof("Scan %q -> %v", kv.Key, vals[col.Name])
		}
	}

	resp.Results = []driver.Result{
		{
			Columns: make([]string, len(desc.Columns)),
			Rows:    rows,
		},
//.........这里部分代码省略.........