// explainDebug fills in four extra debugging values in the current row:
//  - the row index,
//  - the key,
//  - a value string,
//  - a true bool if we are at the end of the row, or a NULL otherwise.
func (n *scanNode) explainDebug(endOfRow bool) {
	if len(n.row) == len(n.visibleCols) {
		n.row = append(n.row, nil, nil, nil, nil)
	}
	debugVals := n.row[len(n.row)-4:]

	debugVals[0] = parser.DInt(n.rowIndex)
	debugVals[1] = parser.DString(n.prettyKey())
	if n.implicitVals != nil {
		debugVals[2] = parser.DString(prettyDatums(n.implicitVals))
	} else {
		// This conversion to DString is odd. `n.explainValue` is already a
		// `Datum`, but logic_test currently expects EXPLAIN DEBUG output
		// to come out formatted using `encodeSQLString`. This is not
		// consistent across all printing of strings in logic_test, though.
		// TODO(tamird/pmattis): figure out a consistent story for string
		// printing in logic_test.
		debugVals[2] = parser.DString(n.explainValue.String())
	}
	if endOfRow {
		debugVals[3] = parser.DBool(true)
		n.rowIndex++
	} else {
		debugVals[3] = parser.DNull
	}
	n.explainValue = nil
}

// ShowColumns of a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW COLUMNS statement.
//          mysql only returns columns you have privileges on.
func (p *planner) ShowColumns(n *parser.ShowColumns) (planNode, *roachpb.Error) {
	desc, pErr := p.getTableDesc(n.Table)
	if pErr != nil {
		return nil, pErr
	}
	v := &valuesNode{
		columns: []column{
			{name: "Field", typ: parser.DummyString},
			{name: "Type", typ: parser.DummyString},
			{name: "Null", typ: parser.DummyBool},
			{name: "Default", typ: parser.DummyString},
		},
	}
	for i, col := range desc.Columns {
		defaultExpr := parser.Datum(parser.DNull)
		if e := desc.Columns[i].DefaultExpr; e != nil {
			defaultExpr = parser.DString(*e)
		}
		v.rows = append(v.rows, []parser.Datum{
			parser.DString(desc.Columns[i].Name),
			parser.DString(col.Type.SQLString()),
			parser.DBool(desc.Columns[i].Nullable),
			defaultExpr,
		})
	}
	return v, nil
}

func (n *scanNode) explainDebug(endOfRow, outputRow bool) {
	if n.row == nil {
		n.row = make([]parser.Datum, len(n.columns))
	}
	n.row[0] = parser.DInt(n.rowIndex)
	n.row[1] = parser.DString(n.prettyKey())
	if n.implicitVals != nil {
		n.row[2] = parser.DString(prettyKeyVals(n.implicitVals))
	} else {
		// This conversion to DString is odd. `n.explainValue` is already a
		// `Datum`, but logic_test currently expects EXPLAIN DEBUG output
		// to come out formatted using `encodeSQLString`. This is not
		// consistent across all printing of strings in logic_test, though.
		// TODO(tamird/pmattis): figure out a consistent story for string
		// printing in logic_test.
		n.row[2] = parser.DString(n.explainValue.String())
	}
	if endOfRow {
		n.row[3] = parser.DBool(outputRow)
		n.rowIndex++
	} else {
		n.row[3] = parser.DNull
	}
	n.explainValue = nil
}

// ShowIndex returns all the indexes for a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW INDEX statement.
//          mysql requires some privilege for any column.
func (p *planner) ShowIndex(n *parser.ShowIndex) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}

	v := &valuesNode{columns: []string{"Table", "Name", "Unique", "Seq", "Column", "Storing"}}

	name := n.Table.Table()
	for _, index := range append([]IndexDescriptor{desc.PrimaryIndex}, desc.Indexes...) {
		j := 1
		for i, cols := range [][]string{index.ColumnNames, index.StoreColumnNames} {
			for _, col := range cols {
				v.rows = append(v.rows, []parser.Datum{
					parser.DString(name),
					parser.DString(index.Name),
					parser.DBool(index.Unique),
					parser.DInt(j),
					parser.DString(col),
					parser.DBool(i == 1),
				})
				j++
			}
		}
	}
	return v, nil
}

// Arg implements the Args interface
func (p parameters) Arg(name string) (parser.Datum, bool) {
	if !unicode.IsDigit(rune(name[0])) {
		// TODO(pmattis): Add support for named parameters (vs the numbered
		// parameter support below).
		return nil, false
	}
	i, err := strconv.ParseInt(name, 10, 0)
	if err != nil {
		return nil, false
	}
	if i < 1 || int(i) > len(p) {
		return nil, false
	}
	arg := p[i-1].GetValue()
	if arg == nil {
		return parser.DNull, true
	}
	switch t := arg.(type) {
	case *bool:
		return parser.DBool(*t), true
	case *int64:
		return parser.DInt(*t), true
	case *float64:
		return parser.DFloat(*t), true
	case []byte:
		return parser.DString(t), true
	case *string:
		return parser.DString(*t), true
	default:
		panic(fmt.Sprintf("unexpected type %T", t))
	}
}

func (vals *debugValues) AsRow() parser.DTuple {
	keyVal := parser.DNull
	if vals.key != "" {
		keyVal = parser.DString(vals.key)
	}

	// The "output" value is NULL for partial rows, or a DBool indicating if the row passed the
	// filtering.
	outputVal := parser.DNull

	switch vals.output {
	case debugValueFiltered:
		outputVal = parser.DBool(false)

	case debugValueRow:
		outputVal = parser.DBool(true)
	}

	return parser.DTuple{
		parser.DInt(vals.rowIdx),
		keyVal,
		parser.DString(vals.value),
		outputVal,
	}
}

// Show a session-local variable name.
func (p *planner) Show(n *parser.Show) (planNode, *roachpb.Error) {
	name := strings.ToUpper(n.Name)

	v := &valuesNode{columns: []ResultColumn{{Name: name, Typ: parser.DummyString}}}

	switch name {
	case `DATABASE`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.session.Database)})
	case `TIME ZONE`:
		loc, err := p.evalCtx.GetLocation()
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		v.rows = append(v.rows, []parser.Datum{parser.DString(loc.String())})
	case `SYNTAX`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(parser.Syntax(p.session.Syntax).String())})
	case `TRANSACTION ISOLATION LEVEL`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.txn.Proto.Isolation.String())})
	case `TRANSACTION PRIORITY`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.txn.UserPriority.String())})
	default:
		return nil, roachpb.NewUErrorf("unknown variable: %q", name)
	}

	return v, nil
}

func (n *explainTraceNode) Next() bool {
	first := n.rows == nil
	if first {
		n.rows = []parser.DTuple{}
	}
	for !n.exhausted && len(n.rows) <= 1 {
		var vals debugValues
		if !n.plan.Next() {
			n.exhausted = true
			if pErr := n.PErr(); pErr != nil {
				n.txn.Trace.LogEvent(pErr.GoError().Error())
			}
			n.txn.Trace.LogEvent("tracing completed")
			n.txn.Trace.Finish()
			n.txn.Trace = nil
		} else {
			vals = n.plan.DebugValues()
		}
		var basePos int
		if len(n.txn.CollectedSpans) == 0 {
			if !n.exhausted {
				n.txn.CollectedSpans = append(n.txn.CollectedSpans, basictracer.RawSpan{
					Context: basictracer.Context{},
					Logs:    []opentracing.LogData{{Timestamp: n.lastTS}},
				})
			}
			basePos = n.lastPos + 1
		}

		for _, sp := range n.txn.CollectedSpans {
			for i, entry := range sp.Logs {
				var timeVal string
				if i > 0 {
					timeVal = time.Duration(entry.Timestamp.Sub(n.lastTS)).String()
				}

				n.rows = append(n.rows, append(parser.DTuple{
					parser.DTimestamp{Time: entry.Timestamp},
					parser.DString(timeVal),
					parser.DInt(basePos + i),
					parser.DString(sp.Operation),
					parser.DString(entry.Event),
				}, vals.AsRow()...))
				n.lastTS, n.lastPos = entry.Timestamp, i
			}
		}
		n.txn.CollectedSpans = nil
	}

	if first {
		return len(n.rows) > 0
	}
	if len(n.rows) <= 1 {
		return false
	}
	n.rows = n.rows[1:]
	return true
}

// SetUIData is an endpoint that sets the data associated with a key.
func (s *adminServer) SetUIData(_ context.Context, req *SetUIDataRequest) (*SetUIDataResponse, error) {
	if len(req.Key) == 0 {
		return nil, grpc.Errorf(codes.InvalidArgument, "key cannot be empty")
	}

	var session sql.Session
	user := s.getUser(req)

	// Do an upsert of the key.
	br := s.sqlExecutor.ExecuteStatements(user, &session, "BEGIN;", nil)
	if err := s.checkQueryResults(br.ResultList, 1); err != nil {
		return nil, s.serverError(err)
	}

	// See if the key already exists.
	alreadyExists := true
	if _, _, err := s.getUIData(&session, user, req.Key); err != nil {
		if err != errUIKeyNotFound {
			return nil, s.serverError(err)
		}
		alreadyExists = false
	}

	// INSERT or UPDATE as appropriate.
	ts := session.Txn.TxnTimestamp
	if alreadyExists {
		query := "UPDATE system.ui SET value = $1, lastUpdated = $2 WHERE key = $3; COMMIT;"
		params := []parser.Datum{
			parser.DString(req.Value),            // $1
			parser.DTimestamp{Time: ts.GoTime()}, // $2
			parser.DString(req.Key),              // $3
		}
		r := s.sqlExecutor.ExecuteStatements(user, &session, query, params)
		if err := s.checkQueryResults(r.ResultList, 2); err != nil {
			return nil, s.serverError(err)
		}
		if a, e := r.ResultList[0].RowsAffected, 1; a != e {
			return nil, s.serverErrorf("rows affected %d != expected %d", a, e)
		}
	} else {
		query := "INSERT INTO system.ui (key, value, lastUpdated) VALUES ($1, $2, $3); COMMIT;"
		params := []parser.Datum{
			parser.DString(req.Key),              // $1
			parser.DBytes(req.Value),             // $2
			parser.DTimestamp{Time: ts.GoTime()}, // $3
		}
		r := s.sqlExecutor.ExecuteStatements(user, &session, query, params)
		if err := s.checkQueryResults(r.ResultList, 2); err != nil {
			return nil, s.serverError(err)
		}
		if a, e := r.ResultList[0].RowsAffected, 1; a != e {
			return nil, s.serverErrorf("rows affected %d != expected %d", a, e)
		}
	}

	return &SetUIDataResponse{}, nil
}

// SetUIData is an endpoint that stores the given key/value pairs in the
// system.ui table. See GetUIData for more details on semantics.
func (s *adminServer) SetUIData(ctx context.Context, req *SetUIDataRequest) (*SetUIDataResponse, error) {
	if len(req.KeyValues) == 0 {
		return nil, grpc.Errorf(codes.InvalidArgument, "KeyValues cannot be empty")
	}

	session := sql.NewSession(sql.SessionArgs{User: s.getUser(req)}, s.sqlExecutor, nil)

	for key, val := range req.KeyValues {
		// Do an upsert of the key. We update each key in a separate transaction to
		// avoid long-running transactions and possible deadlocks.
		br := s.sqlExecutor.ExecuteStatements(ctx, session, "BEGIN;", nil)
		if err := s.checkQueryResults(br.ResultList, 1); err != nil {
			return nil, s.serverError(err)
		}

		// See if the key already exists.
		resp, err := s.getUIData(session, s.getUser(req), []string{key})
		if err != nil {
			return nil, s.serverError(err)
		}
		_, alreadyExists := resp.KeyValues[key]

		// INSERT or UPDATE as appropriate.
		if alreadyExists {
			query := "UPDATE system.ui SET value = $1, lastUpdated = NOW() WHERE key = $2; COMMIT;"
			params := []parser.Datum{
				parser.DString(val), // $1
				parser.DString(key), // $2
			}
			r := s.sqlExecutor.ExecuteStatements(ctx, session, query, params)
			if err := s.checkQueryResults(r.ResultList, 2); err != nil {
				return nil, s.serverError(err)
			}
			if a, e := r.ResultList[0].RowsAffected, 1; a != e {
				return nil, s.serverErrorf("rows affected %d != expected %d", a, e)
			}
		} else {
			query := "INSERT INTO system.ui (key, value, lastUpdated) VALUES ($1, $2, NOW()); COMMIT;"
			params := []parser.Datum{
				parser.DString(key), // $1
				parser.DBytes(val),  // $2
			}
			r := s.sqlExecutor.ExecuteStatements(ctx, session, query, params)
			if err := s.checkQueryResults(r.ResultList, 2); err != nil {
				return nil, s.serverError(err)
			}
			if a, e := r.ResultList[0].RowsAffected, 1; a != e {
				return nil, s.serverErrorf("rows affected %d != expected %d", a, e)
			}
		}
	}

	return &SetUIDataResponse{}, nil
}

func populateExplain(v *valuesNode, plan planNode, level int) {
	name, description, children := plan.ExplainPlan()

	row := parser.DTuple{
		parser.DInt(level),
		parser.DString(name),
		parser.DString(description),
	}
	v.rows = append(v.rows, row)

	for _, child := range children {
		populateExplain(v, child, level+1)
	}
}

func (n *scanNode) explainDebug(endOfRow, outputRow bool) {
	if n.row == nil {
		n.row = make([]parser.Datum, len(n.columns))
	}
	n.row[0] = parser.DInt(n.rowIndex)
	n.row[1] = parser.DString(n.prettyKey())
	n.row[2] = parser.DString(n.explainValue.String())
	if endOfRow {
		n.row[3] = parser.DBool(outputRow)
		n.rowIndex++
	} else {
		n.row[3] = parser.DNull
	}
	n.explainValue = nil
}

// ShowColumns of a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW COLUMNS statement.
//          mysql only returns columns you have privileges on.
func (p *planner) ShowColumns(n *parser.ShowColumns) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	v := &valuesNode{columns: []string{"Field", "Type", "Null"}}
	for i, col := range desc.Columns {
		v.rows = append(v.rows, []parser.Datum{
			parser.DString(desc.Columns[i].Name),
			parser.DString(col.Type.SQLString()),
			parser.DBool(desc.Columns[i].Nullable),
		})
	}
	return v, nil
}

func (n *explainDebugNode) Values() parser.DTuple {
	vals := n.plan.DebugValues()

	keyVal := parser.DNull
	if vals.key != "" {
		keyVal = parser.DString(vals.key)
	}

	return parser.DTuple{
		parser.DInt(vals.rowIdx),
		keyVal,
		parser.DString(vals.value),
		parser.DString(vals.output.String()),
	}
}

// Show a session-local variable name.
func (p *planner) Show(n *parser.Show) (planNode, error) {
	name := strings.ToUpper(n.Name)

	v := &valuesNode{columns: []string{name}}

	switch name {
	case `DATABASE`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.session.Database)})
	case `SYNTAX`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(parser.Syntax(p.session.Syntax).String())})
	default:
		return nil, fmt.Errorf("unknown variable: %q", name)
	}

	return v, nil
}

func (n *scanNode) unmarshalValue(kv client.KeyValue) (parser.Datum, bool) {
	kind, ok := n.colKind[n.colID]
	if !ok {
		n.err = fmt.Errorf("column-id \"%d\" does not exist", n.colID)
		return nil, false
	}
	if kv.Exists() {
		switch kind {
		case ColumnType_INT:
			return parser.DInt(kv.ValueInt()), true
		case ColumnType_BOOL:
			return parser.DBool(kv.ValueInt() != 0), true
		case ColumnType_FLOAT:
			return parser.DFloat(math.Float64frombits(uint64(kv.ValueInt()))), true
		case ColumnType_STRING, ColumnType_BYTES:
			return parser.DString(kv.ValueBytes()), true
		case ColumnType_DATE:
			var t time.Time
			if err := t.UnmarshalBinary(kv.ValueBytes()); err != nil {
				return nil, false
			}
			return parser.DDate{Time: t}, true
		case ColumnType_TIMESTAMP:
			var t time.Time
			if err := t.UnmarshalBinary(kv.ValueBytes()); err != nil {
				return nil, false
			}
			return parser.DTimestamp{Time: t}, true
		case ColumnType_INTERVAL:
			return parser.DInterval{Duration: time.Duration(kv.ValueInt())}, true
		}
	}
	return parser.DNull, true
}

// getUIData returns the value and timestamp for the given UI key. Returns
// errUIKeyNotFound if the key was not found.
func (s *adminServer) getUIData(session *sql.Session, user, key string) ([]byte, GetUIDataResponse_Timestamp, error) {
	zeroTimestamp := GetUIDataResponse_Timestamp{}

	// Query database.
	query := "SELECT value, lastUpdated FROM system.ui WHERE key = $1"
	params := []parser.Datum{parser.DString(key)}
	r := s.sqlExecutor.ExecuteStatements(user, session, query, params)
	if err := s.checkQueryResults(r.ResultList, 1); err != nil {
		return nil, zeroTimestamp, s.serverError(err)
	}
	if len(r.ResultList[0].Rows) == 0 {
		return nil, zeroTimestamp, errUIKeyNotFound
	}

	// Marshal results.
	row := r.ResultList[0].Rows[0]
	dBytes, ok := row.Values[0].(parser.DBytes)
	if !ok {
		return nil, zeroTimestamp, s.serverErrorf("unexpected type for UI value: %T", row.Values[0])
	}
	dTS, ok := row.Values[1].(parser.DTimestamp)
	if !ok {
		return nil, zeroTimestamp,
			s.serverErrorf("unexpected type for UI lastUpdated: %T", row.Values[1])
	}
	nanos := dTS.UnixNano()
	ts := GetUIDataResponse_Timestamp{nanos / 1e9, uint32(nanos % 1e9)}
	return []byte(dBytes), ts, nil
}

func datumFromProto(d driver.Datum) parser.Datum {
	arg := d.Payload
	if arg == nil {
		return parser.DNull
	}
	switch t := arg.(type) {
	case *driver.Datum_BoolVal:
		return parser.DBool(t.BoolVal)
	case *driver.Datum_IntVal:
		return parser.DInt(t.IntVal)
	case *driver.Datum_FloatVal:
		return parser.DFloat(t.FloatVal)
	case *driver.Datum_DecimalVal:
		dec, err := decimal.NewFromString(t.DecimalVal)
		if err != nil {
			panic(fmt.Sprintf("could not parse decimal: %v", err))
		}
		return parser.DDecimal{Decimal: dec}
	case *driver.Datum_BytesVal:
		return parser.DBytes(t.BytesVal)
	case *driver.Datum_StringVal:
		return parser.DString(t.StringVal)
	case *driver.Datum_DateVal:
		return parser.DDate(t.DateVal)
	case *driver.Datum_TimeVal:
		return parser.DTimestamp{Time: t.TimeVal.GoTime()}
	case *driver.Datum_IntervalVal:
		return parser.DInterval{Duration: time.Duration(t.IntervalVal)}
	default:
		panic(fmt.Sprintf("unexpected type %T", t))
	}
}

func decodeTableKey(valType parser.Datum, key []byte) (parser.Datum, []byte, error) {
	var isNull bool
	if key, isNull = encoding.DecodeIfNull(key); isNull {
		return parser.DNull, key, nil
	}
	switch valType.(type) {
	case parser.DBool:
		rkey, i, err := encoding.DecodeVarint(key)
		return parser.DBool(i != 0), rkey, err
	case parser.DInt:
		rkey, i, err := encoding.DecodeVarint(key)
		return parser.DInt(i), rkey, err
	case parser.DFloat:
		rkey, f, err := encoding.DecodeFloat(key, nil)
		return parser.DFloat(f), rkey, err
	case parser.DString:
		rkey, r, err := encoding.DecodeString(key, nil)
		return parser.DString(r), rkey, err
	case parser.DBytes:
		rkey, r, err := encoding.DecodeString(key, nil)
		return parser.DBytes(r), rkey, err
	case parser.DDate:
		rkey, t, err := encoding.DecodeTime(key)
		return parser.DDate{Time: t}, rkey, err
	case parser.DTimestamp:
		rkey, t, err := encoding.DecodeTime(key)
		return parser.DTimestamp{Time: t}, rkey, err
	case parser.DInterval:
		rkey, d, err := encoding.DecodeVarint(key)
		return parser.DInterval{Duration: time.Duration(d)}, rkey, err
	default:
		return nil, nil, util.Errorf("TODO(pmattis): decoded index key: %s", valType.Type())
	}
}

// Arg implements the parser.Args interface.
func (p parameters) Arg(name string) (parser.Datum, bool) {
	i, err := processPositionalArgument(name)
	if err != nil {
		return nil, false
	}
	if i < 1 || int(i) > len(p) {
		return nil, false
	}
	arg := p[i-1].Payload
	if arg == nil {
		return parser.DNull, true
	}
	switch t := arg.(type) {
	case *driver.Datum_BoolVal:
		return parser.DBool(t.BoolVal), true
	case *driver.Datum_IntVal:
		return parser.DInt(t.IntVal), true
	case *driver.Datum_FloatVal:
		return parser.DFloat(t.FloatVal), true
	case *driver.Datum_BytesVal:
		return parser.DBytes(t.BytesVal), true
	case *driver.Datum_StringVal:
		return parser.DString(t.StringVal), true
	case *driver.Datum_DateVal:
		return parser.DDate(t.DateVal), true
	case *driver.Datum_TimeVal:
		return parser.DTimestamp{Time: t.TimeVal.GoTime()}, true
	case *driver.Datum_IntervalVal:
		return parser.DInterval{Duration: time.Duration(t.IntervalVal)}, true
	default:
		panic(fmt.Sprintf("unexpected type %T", t))
	}
}