// 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
}

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,
	}
}

func checkEquivExpr(a, b parser.Expr, qvals qvalMap) error {
	// The expressions above only use the values 1 and 2. Verify that the
	// simplified expressions evaluate to the same value as the original
	// expression for interesting values.
	zero := parser.DInt(0)
	for _, v := range []parser.Datum{zero, zero + 1, zero + 2, zero + 3, parser.DNull} {
		for _, q := range qvals {
			q.datum = v
		}
		da, err := a.Eval(parser.EvalContext{})
		if err != nil {
			return fmt.Errorf("%s: %v", a, err)
		}
		db, err := b.Eval(parser.EvalContext{})
		if err != nil {
			return fmt.Errorf("%s: %v", b, err)
		}
		// This is tricky: we don't require the expressions to produce identical
		// results, but to either both return true or both return not true (either
		// false or NULL).
		if (da == parser.DBool(true)) != (db == parser.DBool(true)) {
			return fmt.Errorf("%s: %s: expected %s, but found %s", a, v, da, db)
		}
	}
	return nil
}

func makeNot(expr parser.Expr) parser.Expr {
	if expr == parser.DBool(true) {
		return parser.DBool(false)
	}
	if expr == parser.DBool(false) {
		return parser.DBool(true)
	}
	return &parser.NotExpr{Expr: expr}
}

func simplifyNotExpr(n *parser.NotExpr) parser.Expr {
	switch t := n.Expr.(type) {
	case *parser.ComparisonExpr:
		op := t.Operator
		switch op {
		case parser.EQ:
			op = parser.NE
		case parser.NE:
			op = parser.EQ
		case parser.GT:
			op = parser.LE
		case parser.GE:
			op = parser.LT
		case parser.LT:
			op = parser.GE
		case parser.LE:
			op = parser.GT
		case parser.In:
			op = parser.NotIn
		case parser.NotIn:
			op = parser.In
		case parser.Like:
			op = parser.NotLike
		case parser.NotLike:
			op = parser.Like
		case parser.SimilarTo:
			op = parser.NotSimilarTo
		case parser.NotSimilarTo:
			op = parser.SimilarTo
		default:
			return parser.DBool(true)
		}
		return simplifyExpr(&parser.ComparisonExpr{
			Operator: op,
			Left:     t.Left,
			Right:    t.Right,
		})

	case *parser.AndExpr:
		// De Morgan's Law: NOT (a AND b) -> (NOT a) OR (NOT b)
		return simplifyExpr(&parser.OrExpr{
			Left:  &parser.NotExpr{Expr: t.Left},
			Right: &parser.NotExpr{Expr: t.Right},
		})

	case *parser.OrExpr:
		// De Morgan's Law: NOT (a OR b) -> (NOT a) AND (NOT b)
		return simplifyExpr(&parser.AndExpr{
			Left:  &parser.NotExpr{Expr: t.Left},
			Right: &parser.NotExpr{Expr: t.Right},
		})
	}
	return parser.DBool(true)
}

func makeOr(left parser.Expr, right parser.Expr) parser.Expr {
	if left == parser.DBool(true) || right == parser.DBool(true) {
		return parser.DBool(true)
	}
	if left == parser.DBool(false) {
		return right
	}
	if right == parser.DBool(false) {
		return left
	}
	return &parser.OrExpr{Left: left, Right: right}
}

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

	desc, err := p.mustGetTableDesc(tn)
	if err != nil {
		return nil, err
	}
	if err := p.anyPrivilege(desc); err != nil {
		return nil, err
	}

	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Table", Typ: parser.TypeString},
			{Name: "Name", Typ: parser.TypeString},
			{Name: "Unique", Typ: parser.TypeBool},
			{Name: "Seq", Typ: parser.TypeInt},
			{Name: "Column", Typ: parser.TypeString},
			{Name: "Direction", Typ: parser.TypeString},
			{Name: "Storing", Typ: parser.TypeBool},
		},
	}

	appendRow := func(index sqlbase.IndexDescriptor, colName string, sequence int,
		direction string, isStored bool) {
		v.rows = append(v.rows, []parser.Datum{
			parser.NewDString(tn.Table()),
			parser.NewDString(index.Name),
			parser.MakeDBool(parser.DBool(index.Unique)),
			parser.NewDInt(parser.DInt(sequence)),
			parser.NewDString(colName),
			parser.NewDString(direction),
			parser.MakeDBool(parser.DBool(isStored)),
		})
	}

	for _, index := range append([]sqlbase.IndexDescriptor{desc.PrimaryIndex}, desc.Indexes...) {
		sequence := 1
		for i, col := range index.ColumnNames {
			appendRow(index, col, sequence, index.ColumnDirections[i].String(), false)
			sequence++
		}
		for _, col := range index.StoreColumnNames {
			appendRow(index, col, sequence, "N/A", true)
			sequence++
		}
	}
	return v, nil
}

// splitFilter splits a boolean expression E into two boolean expressions RES and REM such that:
//
//  - RES contains only variables known to the conversion function (it is "restricted" to a
//    particular set of variables). These variables are also converted as returned by conv.
//
//  - the original expression is equivalent to the conjunction (AND) between the RES and REM
//    expressions.
//
// Splitting allows us to do filtering at various layers, where one layer only knows the values of
// some variables. Instead of evaluating E in an upper layer, we evaluate RES in a lower layer
// and then evaluate REM in the upper layer (on results that passed the RES filter).
//
// Notes:
//  - the implementation is best-effort (it tries to get as much of the expression into RES as
//    possible, and make REM as small as possible).
//  - the original expression is modified in-place and should not be used again.
func splitFilter(expr parser.Expr, conv varConvertFunc) (restricted, remainder parser.Expr) {
	if expr == nil {
		return nil, nil
	}
	restricted, remainder = splitBoolExpr(expr, conv, true)
	if restricted == parser.DBool(true) {
		restricted = nil
	}
	if remainder == parser.DBool(true) {
		remainder = nil
	}
	return restricted, remainder
}

func simplifyComparisonExpr(n *parser.ComparisonExpr) parser.Expr {
	// NormalizeExpr will have left comparisons in the form "<var> <op>
	// <datum>" unless they could not be simplified further in which case
	// simplifyExpr cannot handle them. For example, "lower(a) = 'foo'"
	if isVar(n.Left) && isDatum(n.Right) {
		// All of the comparison operators have the property that when comparing to
		// NULL they evaulate to NULL (see evalComparisonOp). NULL is not the same
		// as false, but in the context of a WHERE clause, NULL is considered
		// not-true which is the same as false.
		if n.Right == parser.DNull {
			return parser.DBool(false)
		}

		switch n.Operator {
		case parser.EQ, parser.NE, parser.GT, parser.GE, parser.LT, parser.LE:
			return n
		case parser.In, parser.NotIn:
			tuple, ok := n.Right.(parser.DTuple)
			if !ok {
				break
			}
			if !typeCheckTuple(n.Left, tuple) {
				break
			}
			sort.Sort(tuple)
			tuple = uniqTuple(tuple)
			if len(tuple) == 0 {
				return parser.DBool(false)
			}
			n.Right = tuple
			return n
		case parser.Like:
			// a LIKE 'foo%' -> a >= "foo" AND a < "fop"
			if d, ok := n.Right.(parser.DString); ok {
				if i := strings.IndexAny(string(d), "_%"); i >= 0 {
					return makePrefixRange(d[:i], n.Left, false)
				}
				return makePrefixRange(d, n.Left, true)
			}
		case parser.SimilarTo:
			// a SIMILAR TO "foo.*" -> a >= "foo" AND a < "fop"
			if d, ok := n.Right.(parser.DString); ok {
				if re, err := regexp.Compile(string(d)); err == nil {
					prefix, complete := re.LiteralPrefix()
					return makePrefixRange(parser.DString(prefix), n.Left, complete)
				}
			}
		}
	}
	return parser.DBool(true)
}

func (v *applyConstraintsVisitor) VisitPost(expr parser.Expr) parser.Expr {
	switch t := expr.(type) {
	case *parser.AndExpr:
		if t.Left == parser.DBool(true) && t.Right == parser.DBool(true) {
			return parser.DBool(true)
		} else if t.Left == parser.DBool(true) {
			return t.Right
		} else if t.Right == parser.DBool(true) {
			return t.Left
		}
	}

	return expr
}

func simplifyOneOrInExpr(left, right *parser.ComparisonExpr) (parser.Expr, parser.Expr) {
	if left.Operator != parser.In && right.Operator != parser.In {
		panic(fmt.Sprintf("IN expression required: %s vs %s", left, right))
	}

	switch left.Operator {
	case parser.EQ:
		switch right.Operator {
		case parser.In:
			left, right = right, left
		}
		fallthrough

	case parser.In:
		tuple, ok := left.Right.(parser.DTuple)
		if !ok {
			return parser.DBool(true), nil
		}

		var tuple2 parser.DTuple
		switch right.Operator {
		case parser.EQ:
			rdatum, rok := right.Right.(parser.Datum)
			if !rok {
				return parser.DBool(true), nil
			}
			tuple2 = parser.DTuple{rdatum}
		case parser.In:
			tuple2, ok = right.Right.(parser.DTuple)
			if !ok {
				return parser.DBool(true), nil
			}
		}

		if !typeCheckTuple(left.Left, tuple2) {
			return left, right
		}

		// We keep the tuples for an in expression in sorted order. So now we just
		// merge the two sorted lists.
		return &parser.ComparisonExpr{
			Operator: parser.In,
			Left:     left.Left,
			Right:    mergeSorted(tuple, tuple2),
		}, nil
	}

	return left, right
}

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

// 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
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}
	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Field", Typ: parser.TypeString},
			{Name: "Type", Typ: parser.TypeString},
			{Name: "Null", Typ: parser.TypeBool},
			{Name: "Default", Typ: parser.TypeString},
		},
	}
	for i, col := range desc.Columns {
		defaultExpr := parser.Datum(parser.DNull)
		if e := desc.Columns[i].DefaultExpr; e != nil {
			defaultExpr = parser.NewDString(*e)
		}
		v.rows = append(v.rows, []parser.Datum{
			parser.NewDString(desc.Columns[i].Name),
			parser.NewDString(col.Type.SQLString()),
			parser.MakeDBool(parser.DBool(desc.Columns[i].Nullable)),
			defaultExpr,
		})
	}
	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
}

// applyConstraints applies the constraints on values specified by constraints
// to an expression, simplifying the expression where possible. For example, if
// the expression is "a = 1" and the constraint is "a = 1", the expression can
// be simplified to "true". If the expression is "a = 1 AND b > 2" and the
// constraint is "a = 1", the expression is simplified to "b > 2".
//
// Note that applyConstraints currently only handles simple cases.
func applyConstraints(expr parser.Expr, constraints indexConstraints) parser.Expr {
	v := &applyConstraintsVisitor{}
	for _, c := range constraints {
		v.constraint = c
		expr = parser.WalkExpr(v, expr)
		// We can only continue to apply the constraints if the constraints we have
		// applied so far are equality constraints. There are two cases to
		// consider: the first is that both the start and end constraints are
		// equality.
		if c.start == c.end {
			if c.start.Operator == parser.EQ {
				continue
			}
			// The second case is that both the start and end constraint are an IN
			// operator with only a single value.
			if c.start.Operator == parser.In && len(c.start.Right.(parser.DTuple)) == 1 {
				continue
			}
		}
		break
	}
	if expr == parser.DBool(true) {
		return nil
	}
	return expr
}

// applyConstraints applies the constraints on values specified by constraints
// to an expression, simplifying the expression where possible. For example, if
// the expression is "a = 1" and the constraint is "a = 1", the expression can
// be simplified to "true". If the expression is "a = 1 AND b > 2" and the
// constraint is "a = 1", the expression is simplified to "b > 2".
//
// Note that applyConstraints currently only handles simple cases.
func applyConstraints(expr parser.Expr, constraints orIndexConstraints) parser.Expr {
	if len(constraints) != 1 {
		// We only support simplifying the expressions if there aren't multiple
		// disjunctions (top-level OR branches).
		return expr
	}
	v := &applyConstraintsVisitor{}
	for _, c := range constraints[0] {
		v.constraint = c
		expr, _ = parser.WalkExpr(v, expr)
		// We can only continue to apply the constraints if the constraints we have
		// applied so far are equality constraints. There are two cases to
		// consider: the first is that both the start and end constraints are
		// equality.
		if c.start == c.end {
			if c.start.Operator == parser.EQ {
				continue
			}
			// The second case is that both the start and end constraint are an IN
			// operator with only a single value.
			if c.start.Operator == parser.In && len(c.start.Right.(parser.DTuple)) == 1 {
				continue
			}
		}
		break
	}
	if expr == parser.DBool(true) {
		return nil
	}
	return expr
}

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

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:
		var i int64
		key, i = encoding.DecodeVarint(key)
		return parser.DBool(i != 0), key, nil
	case parser.DInt:
		var i int64
		key, i = encoding.DecodeVarint(key)
		return parser.DInt(i), key, nil
	case parser.DFloat:
		var f float64
		key, f = encoding.DecodeFloat(key, nil)
		return parser.DFloat(f), key, nil
	case parser.DString:
		var r string
		key, r = encoding.DecodeString(key, nil)
		return parser.DString(r), key, nil
	default:
		return nil, nil, util.Errorf("TODO(pmattis): decoded index key: %s", valType.Type())
	}
}

// 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
}