// CreatePrimaryIndex implements datastore.Indexer{} interface. Create or
// return a primary index on this keyspace
func (gsi *gsiKeyspace) CreatePrimaryIndex(
	requestId, name string, with value.Value) (datastore.PrimaryIndex, errors.Error) {

	var withJSON []byte
	var err error
	if with != nil {
		if withJSON, err = with.MarshalJSON(); err != nil {
			return nil, errors.NewError(err, "GSI error marshalling WITH clause")
		}
	}
	defnID, err := gsi.gsiClient.CreateIndex(
		name,
		gsi.keyspace,       /*bucket-name*/
		string(c.ForestDB), /*using, by default always forestdb*/
		"N1QL",             /*exprType*/
		"",                 /*partnStr*/
		"",                 /*whereStr*/
		nil,                /*secStrs*/
		true,               /*isPrimary*/
		withJSON)
	if err != nil {
		return nil, errors.NewError(err, "GSI CreatePrimaryIndex()")
	}
	// refresh to get back the newly created index.
	if err := gsi.Refresh(); err != nil {
		return nil, err
	}
	index, errr := gsi.IndexById(defnID2String(defnID))
	if errr != nil {
		return nil, errr
	}
	return index.(datastore.PrimaryIndex), nil
}

/*
This method removes all the occurences of the second value from the
first array value.
*/
func (this *ArrayRemove) Apply(context Context, first, second value.Value) (value.Value, error) {
	if first.Type() == value.MISSING {
		return first, nil
	}

	if first.Type() != value.ARRAY {
		return value.NULL_VALUE, nil
	}

	if second.Type() <= value.NULL {
		return first, nil
	}

	fa := first.Actual().([]interface{})
	if len(fa) == 0 {
		return first, nil
	}

	ra := make([]interface{}, 0, len(fa))
	for _, f := range fa {
		if !second.Equals(value.NewValue(f)).Truth() {
			ra = append(ra, f)
		}
	}

	return value.NewValue(ra), nil
}

/*
Perform either case-sensitive or case-insensitive field lookup.
*/
func (this *Field) Apply(context Context, first, second value.Value) (value.Value, error) {
	switch second.Type() {
	case value.STRING:
		s := second.Actual().(string)
		v, ok := first.Field(s)

		if !ok && this.caseInsensitive {
			s = strings.ToLower(s)
			fields := first.Fields()
			for f, val := range fields {
				if s == strings.ToLower(f) {
					return value.NewValue(val), nil
				}
			}
		}

		return v, nil
	case value.MISSING:
		return value.MISSING_VALUE, nil
	default:
		if first.Type() == value.MISSING {
			return value.MISSING_VALUE, nil
		} else {
			return value.NULL_VALUE, nil
		}
	}
}

/*
This method returns an object value. The input of types
missing, null and object return themselves. For all other
values, return an _EMPTY_OBJECT value.
*/
func (this *ToObject) Apply(context Context, arg value.Value) (value.Value, error) {
	switch arg.Type() {
	case value.MISSING, value.NULL, value.OBJECT:
		return arg, nil
	}

	return _EMPTY_OBJECT, nil
}

/*
Evaluates the Is Missing comparison operation for expressions.
Return true if the input argument value is a missing value,
else return false.
*/
func (this *IsMissing) Apply(context Context, arg value.Value) (value.Value, error) {
	switch arg.Type() {
	case value.MISSING:
		return value.TRUE_VALUE, nil
	default:
		return value.FALSE_VALUE, nil
	}
}

func (this *LE) Apply(context Context, first, second value.Value) (value.Value, error) {
	cmp := first.Compare(second)
	switch actual := cmp.Actual().(type) {
	case float64:
		return value.NewValue(actual <= 0), nil
	}

	return cmp, nil
}

func (this *Base64) Apply(context Context, operand value.Value) (value.Value, error) {
	if operand.Type() == value.MISSING {
		return operand, nil
	}

	bytes, _ := operand.MarshalJSON() // Ignore errors from BINARY values
	str := base64.StdEncoding.EncodeToString(bytes)
	return value.NewValue(str), nil
}

func testObjectRemove(e1, e2 Expression, er value.Value, t *testing.T) {
	eop := NewObjectRemove(e1, e2)
	rv, err := eop.Evaluate(nil, nil)
	if err != nil {
		t.Errorf("received error %v", err)
	}
	if er.Collate(rv) != 0 {
		t.Errorf("mismatch received %v expected %v", rv.Actual(), er.Actual())
	}
}

func testArrayInsert_eval(e1, e2, e3 Expression, er value.Value, t *testing.T) {
	eai := NewArrayInsert(e1, e2, e3)
	rv, err := eai.Evaluate(nil, nil)
	if err != nil {
		t.Errorf("received error %v", err)
	}
	if er.Collate(rv) != 0 {
		t.Errorf("mismatch received %v expected %v", rv.Actual(), er.Actual())
	}
}

/*
Aggregate input partial values into cumulative result value.
If partial result is null return the current cumulative value,
and if the cumulative result is null, return the partial value.
For non null partial and cumulative values, call Collate and
return the smaller value depending on the N1QL collation order.
*/
func (this *Min) cumulatePart(part, cumulative value.Value, context Context) (value.Value, error) {
	if part == value.NULL_VALUE {
		return cumulative, nil
	} else if cumulative == value.NULL_VALUE {
		return part, nil
	} else if part.Collate(cumulative) < 0 {
		return part, nil
	} else {
		return cumulative, nil
	}
}

func (this *ParentScan) RunOnce(context *Context, parent value.Value) {
	this.once.Do(func() {
		defer context.Recover()       // Recover from any panic
		defer close(this.itemChannel) // Broadcast that I have stopped
		defer this.notify()           // Notify that I have stopped

		// Shallow copy of the parent includes
		// correlated and annotated aspects
		this.sendItem(parent.Copy().(value.AnnotatedValue))
	})
}

func (this *Not) Apply(context Context, arg value.Value) (value.Value, error) {
	switch arg.Type() {
	case value.MISSING, value.NULL:
		return arg, nil
	default:
		if arg.Truth() {
			return value.FALSE_VALUE, nil
		} else {
			return value.TRUE_VALUE, nil
		}
	}
}

/*
Aggregates input data by evaluating operands. For missing
item values, return the input value itself. Call
setAdd to compute the intermediate aggregate value
and return it.
*/
func (this *ArrayAggDistinct) CumulateInitial(item, cumulative value.Value, context Context) (value.Value, error) {
	item, e := this.Operand().Evaluate(item, context)
	if e != nil {
		return nil, e
	}

	if item.Type() <= value.MISSING || item.Type() == value.BINARY {
		return cumulative, nil
	}

	return setAdd(item, cumulative)
}

/*
Aggregates input data by evaluating operands. For missing
item values, return the input value itself. Call
cumulatePart to compute the intermediate aggregate value
and return it.
*/
func (this *ArrayAgg) CumulateInitial(item, cumulative value.Value, context Context) (value.Value, error) {
	item, e := this.Operand().Evaluate(item, context)
	if e != nil {
		return nil, e
	}

	if item.Type() <= value.MISSING || item.Type() == value.BINARY {
		return cumulative, nil
	}

	return this.cumulatePart(value.NewValue([]interface{}{item}), cumulative, context)
}

/*
Aggregates input data by evaluating operands.For all
values other than Number, return the input value itself.
Call setAdd to compute the intermediate aggregate value
and return it.
*/
func (this *AvgDistinct) CumulateInitial(item, cumulative value.Value, context Context) (value.Value, error) {
	item, e := this.Operand().Evaluate(item, context)
	if e != nil {
		return nil, e
	}

	if item.Type() != value.NUMBER {
		return cumulative, nil
	}

	return setAdd(item, cumulative)
}

func (this *Sum) CumulateInitial(item, cumulative value.Value, context Context) (value.Value, error) {
	item, e := this.Operand().Evaluate(item, context)
	if e != nil {
		return nil, e
	}

	if item.Type() != value.NUMBER {
		return cumulative, nil
	}

	return this.cumulatePart(item, cumulative, context)
}

func (this *Order) Less(i, j int) bool {
	v1 := this.values[i]
	v2 := this.values[j]

	var ev1, ev2 value.Value
	var c int
	var e error

	for i, term := range this.plan.Terms() {
		s := this.terms[i]

		sv1 := v1.GetAttachment(s)
		switch sv1 := sv1.(type) {
		case value.Value:
			ev1 = sv1
		default:
			ev1, e = term.Expression().Evaluate(v1, this.context)
			if e != nil {
				this.context.Error(errors.NewEvaluationError(e, "ORDER BY"))
				return false
			}

			v1.SetAttachment(s, ev1)
		}

		sv2 := v2.GetAttachment(s)
		switch sv2 := sv2.(type) {
		case value.Value:
			ev2 = sv2
		default:
			ev2, e = term.Expression().Evaluate(v2, this.context)
			if e != nil {
				this.context.Error(errors.NewEvaluationError(e, "ORDER BY"))
				return false
			}

			v2.SetAttachment(s, ev2)
		}

		c = ev1.Collate(ev2)

		if c == 0 {
			continue
		} else if term.Descending() {
			return c > 0
		} else {
			return c < 0
		}
	}

	return false
}

/*
Aggregates input data by evaluating operands. For all
values other than Number, return the input value itself. Call
cumulatePart to compute the intermediate aggregate value
and return it.
*/
func (this *Avg) CumulateInitial(item, cumulative value.Value, context Context) (value.Value, error) {
	item, e := this.Operand().Evaluate(item, context)
	if e != nil {
		return nil, e
	}

	if item.Type() != value.NUMBER {
		return cumulative, nil
	}

	part := value.NewValue(map[string]interface{}{"sum": item.Actual(), "count": 1})
	return this.cumulatePart(part, cumulative, context)
}

/*
This method checks to see if the values of the two input
expressions are equal, and if true then returns a positive
infinity using the math package method Inf(1). If not it
returns the first input value. Use the Equals method for the
two values to determine equality.
*/
func (this *PosInfIf) Apply(context Context, first, second value.Value) (value.Value, error) {
	eq := first.Equals(second)
	switch eq.Type() {
	case value.MISSING, value.NULL:
		return eq, nil
	default:
		if eq.Truth() {
			return _POS_INF_VALUE, nil
		} else {
			return first, nil
		}
	}
}

func (this *Cover) Evaluate(item value.Value, context Context) (value.Value, error) {
	var rv value.Value
	switch item := item.(type) {
	case value.AnnotatedValue:
		rv = item.GetCover(this.text)
	}

	if rv == nil {
		return value.MISSING_VALUE, errors.NewEvaluationError(nil, "cover("+this.text+")")
	}

	return rv, nil
}