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

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

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

// CreateDatabase creates a database if it doesn't exist.
func (s *Server) CreateDatabase(session *Session, p *parser.CreateDatabase, args []sqlwire.Datum, resp *sqlwire.Response) {
	if p.Name == "" {
		resp.SetGoError(errors.New("empty database name"))
		return
	}

	nameKey := keys.MakeNameMetadataKey(structured.RootNamespaceID, strings.ToLower(p.Name))
	if gr, err := s.db.Get(nameKey); err != nil {
		resp.SetGoError(err)
		return
	} else if gr.Exists() {
		if p.IfNotExists {
			return
		}
		resp.SetGoError(fmt.Errorf("database \"%s\" already exists", p.Name))
		return
	}
	ir, err := s.db.Inc(keys.DescIDGenerator, 1)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	nsID := uint32(ir.ValueInt() - 1)
	if err := s.db.CPut(nameKey, nsID, nil); err != nil {
		// TODO(pmattis): Need to handle if-not-exists here as well.
		resp.SetGoError(err)
		return
	}
}

// ShowColumns of a table
func (s *Server) ShowColumns(session *Session, p *parser.ShowColumns, args []sqlwire.Datum, resp *sqlwire.Response) {
	desc, err := s.getTableDesc(session.Database, p.Table)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	var rows []sqlwire.Result_Row
	for i, col := range desc.Columns {
		t := col.Type.SQLString()
		rows = append(rows, sqlwire.Result_Row{
			Values: []sqlwire.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 = []sqlwire.Result{
		{
			Columns: []string{"Field", "Type", "Null"},
			Rows:    rows,
		},
	}
}

// ShowTables returns all the tables.
func (s *Server) ShowTables(session *Session, p *parser.ShowTables, args []sqlwire.Datum, resp *sqlwire.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 []sqlwire.Result_Row

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

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

func echo(sql string, resp *sqlwire.Response) {
	resp.Results = []sqlwire.Result{
		{
			Columns: []string{"echo"},
			Rows: []sqlwire.Result_Row{
				{
					Values: []sqlwire.Datum{
						{
							StringVal: &sql,
						},
					},
				},
			},
		},
	}
}

// ShowTables returns all the tables.
func (s *Server) ShowTables(p *parser.ShowTables, args []sqlwire.Datum, resp *sqlwire.Response) {
	if p.Name == "" {
		if s.database == "" {
			resp.SetGoError(errors.New("no database specified"))
			return
		}
		p.Name = s.database
	}
	dbID, err := s.lookupDatabase(p.Name)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	prefix := keys.MakeNameMetadataKey(dbID, "")
	sr, err := s.db.Scan(prefix, prefix.PrefixEnd(), 0)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	var rows []sqlwire.Result_Row

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

// Select selects rows from a single table.
func (s *Server) Select(session *Session, p *parser.Select, args []sqlwire.Datum, resp *sqlwire.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 []sqlwire.Result_Row
	var primaryKey []byte
	vals := valMap{}
	for _, kv := range sr {
		if primaryKey != nil && !bytes.HasPrefix(kv.Key, primaryKey) {
			if output, err := shouldOutputRow(p.Where, vals); err != nil {
				return err
			} else if output {
				rows = append(rows, outputRow(desc.Columns, vals))
			}
			vals = valMap{}
		}

		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])
		}
	}

	if output, err := shouldOutputRow(p.Where, vals); err != nil {
		return err
	} else if output {
		rows = append(rows, outputRow(desc.Columns, vals))
	}

	resp.Results = []sqlwire.Result{
		{
			Columns: make([]string, len(desc.Columns)),
			Rows:    rows,
		},
	}
	for i, col := range desc.Columns {
		resp.Results[0].Columns[i] = col.Name
	}
	return nil
}

// Update is unimplemented.
func (s *Server) Update(session *Session, p *parser.Update, args []sqlwire.Datum, resp *sqlwire.Response) {

	resp.SetGoError(fmt.Errorf("TODO(pmattis): unimplemented: %T %s", p, p))
}

// Insert inserts rows into the database.
func (s *Server) Insert(session *Session, p *parser.Insert, args []sqlwire.Datum, resp *sqlwire.Response) {

	desc, err := s.getTableDesc(session.Database, p.Table)
	if err != nil {
		resp.SetGoError(err)
		return
	}

	// Determine which columns we're inserting into.
	cols, err := s.processColumns(desc, p.Columns)
	if err != nil {
		resp.SetGoError(err)
		return
	}

	// Construct a map from column ID to the index the value appears at within a
	// row.
	colMap := map[uint32]int{}
	for i, c := range cols {
		colMap[c.ID] = i
	}

	// Verify we have at least the columns that are part of the primary key.
	for i, id := range desc.Indexes[0].ColumnIDs {
		if _, ok := colMap[id]; !ok {
			resp.SetGoError(fmt.Errorf("missing \"%s\" primary key column",
				desc.Indexes[0].ColumnNames[i]))
			return
		}
	}

	// Transform the values into a rows object. This expands SELECT statements or
	// generates rows from the values contained within the query.
	r, err := s.processInsertRows(p.Rows)
	if err != nil {
		resp.SetGoError(err)
		return
	}

	b := &client.Batch{}
	for _, row := range r {
		if len(row.Values) != len(cols) {
			resp.SetGoError(fmt.Errorf("invalid values for columns: %d != %d", len(row.Values), len(cols)))
			return
		}
		indexKey := encodeIndexKeyPrefix(desc.ID, desc.Indexes[0].ID)
		primaryKey, err := encodeIndexKey(desc.Indexes[0], colMap, cols, row.Values, indexKey)
		if err != nil {
			resp.SetGoError(err)
			return
		}
		for i, val := range row.Values {
			key := encodeColumnKey(desc, cols[i], primaryKey)
			if log.V(2) {
				log.Infof("Put %q -> %v", key, val)
			}
			// TODO(pmattis): Need to convert the value type to the column type.
			// TODO(vivek): We need a better way of storing Datum.
			if val.BoolVal != nil {
				b.Put(key, *val.BoolVal)
			} else if val.IntVal != nil {
				b.Put(key, *val.IntVal)
			} else if val.UintVal != nil {
				b.Put(key, *val.UintVal)
			} else if val.FloatVal != nil {
				b.Put(key, *val.FloatVal)
			} else if val.BytesVal != nil {
				b.Put(key, val.BytesVal)
			} else if val.StringVal != nil {
				b.Put(key, *val.StringVal)
			}
		}
	}
	if err := s.db.Run(b); err != nil {
		resp.SetGoError(err)
		return
	}
}

// CreateTable creates a table if it doesn't already exist.
func (s *Server) CreateTable(session *Session, p *parser.CreateTable, args []sqlwire.Datum, resp *sqlwire.Response) {
	if err := s.normalizeTableName(session.Database, p.Table); err != nil {
		resp.SetGoError(err)
		return
	}

	dbID, err := s.lookupDatabase(p.Table.Qualifier)
	if err != nil {
		resp.SetGoError(err)
		return
	}

	desc, err := makeTableDesc(p)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	if err := desc.AllocateIDs(); err != nil {
		resp.SetGoError(err)
		return
	}

	nameKey := keys.MakeNameMetadataKey(dbID, p.Table.Name)

	// This isn't strictly necessary as the conditional put below will fail if
	// the key already exists, but it seems good to avoid the table ID allocation
	// in most cases when the table already exists.
	if gr, err := s.db.Get(nameKey); err != nil {
		resp.SetGoError(err)
		return
	} else if gr.Exists() {
		if p.IfNotExists {
			return
		}
		resp.SetGoError(fmt.Errorf("table \"%s\" already exists", p.Table))
		return
	}

	ir, err := s.db.Inc(keys.DescIDGenerator, 1)
	if err != nil {
		resp.SetGoError(err)
		return
	}
	desc.ID = uint32(ir.ValueInt() - 1)

	// TODO(pmattis): Be cognizant of error messages when this is ported to the
	// server. The error currently returned below is likely going to be difficult
	// to interpret.
	err = s.db.Txn(func(txn *client.Txn) error {
		descKey := keys.MakeDescMetadataKey(desc.ID)
		b := &client.Batch{}
		b.CPut(nameKey, descKey, nil)
		b.Put(descKey, &desc)
		return txn.Commit(b)
	})
	if err != nil {
		// TODO(pmattis): Need to handle if-not-exists here as well.
		resp.SetGoError(err)
		return
	}
}

// Delete is unimplemented.
func (s *Server) Delete(p *parser.Delete, args []sqlwire.Datum, resp *sqlwire.Response) {
	resp.SetGoError(fmt.Errorf("TODO(pmattis): unimplemented: %T %s", p, p))
}