/*
Qualify all identifiers for the parent expression. Checks for
duplicate aliases.
*/
func (this *KeyspaceTerm) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	keyspace := this.Alias()
	if keyspace == "" {
		err = errors.NewNoTermNameError("FROM", "plan.keyspace.requires_name_or_alias")
		return
	}

	if this.keys != nil {
		_, err = this.keys.Accept(parent)
		if err != nil {
			return
		}
	}

	_, ok := parent.Allowed.Field(keyspace)
	if ok {
		err = errors.NewDuplicateAliasError("subquery", keyspace, "plan.keyspace.duplicate_alias")
		return nil, err
	}

	allowed := value.NewScopeValue(make(map[string]interface{}), parent.Allowed)
	allowed.SetField(keyspace, keyspace)

	f = expression.NewFormalizer()
	f.Keyspace = keyspace
	f.Allowed = allowed
	return
}

func buildFor(f *algebra.UpdateFor, val value.Value, arrays []value.Value, context *Context) ([]value.Value, error) {
	n := -1
	for _, a := range arrays {
		act := a.Actual()
		switch act := act.(type) {
		case []interface{}:
			if n < 0 || len(act) < n {
				n = len(act)
			}
		}
	}

	if n < 0 {
		return nil, nil
	}

	rv := make([]value.Value, n)
	for i, _ := range rv {
		rv[i] = value.NewScopeValue(make(map[string]interface{}, len(f.Bindings())), val)
		for j, b := range f.Bindings() {
			v, ok := arrays[j].Index(i)
			if ok {
				rv[i].SetField(b.Variable(), v)
			}
		}
	}

	return rv, nil
}

/*
Qualify all identifiers for the parent expression. Checks for
duplicate aliases.
*/
func (this *SubqueryTerm) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	err = this.subquery.Formalize()
	if err != nil {
		return
	}

	alias := this.Alias()
	if alias == "" {
		err = errors.NewNoTermNameError("FROM", "plan.subquery.requires_name_or_alias")
		return
	}

	_, ok := parent.Allowed.Field(alias)
	if ok {
		err = errors.NewDuplicateAliasError("subquery", alias, "plan.subquery.duplicate_alias")
		return nil, err
	}

	allowed := value.NewScopeValue(make(map[string]interface{}), parent.Allowed)
	allowed.SetField(alias, alias)

	f = expression.NewFormalizer()
	f.Keyspace = alias
	f.Allowed = allowed
	return
}

func (this *KeyScan) 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

		keys, e := this.plan.Keys().Evaluate(parent, context)
		if e != nil {
			context.Error(errors.NewEvaluationError(e, "KEYS"))
			return
		}

		actuals := keys.Actual()
		switch actuals.(type) {
		case []interface{}:
		case nil:
			actuals = []interface{}(nil)
		default:
			actuals = []interface{}{actuals}
		}

		acts := actuals.([]interface{})

		for _, key := range acts {
			cv := value.NewScopeValue(make(map[string]interface{}), parent)
			av := value.NewAnnotatedValue(cv)
			av.SetAttachment("meta", map[string]interface{}{"id": key})
			if !this.sendItem(av) {
				return
			}
		}
	})
}

func (this *Any) Evaluate(item value.Value, context Context) (value.Value, error) {
	missing := false
	null := false

	barr := make([][]interface{}, len(this.bindings))
	for i, b := range this.bindings {
		bv, err := b.Expression().Evaluate(item, context)
		if err != nil {
			return nil, err
		}

		if b.Descend() {
			buffer := make([]interface{}, 0, 256)
			bv = value.NewValue(bv.Descendants(buffer))
		}

		switch bv.Type() {
		case value.ARRAY:
			barr[i] = bv.Actual().([]interface{})
		case value.MISSING:
			missing = true
		default:
			null = true
		}
	}

	if missing {
		return value.MISSING_VALUE, nil
	}

	if null {
		return value.NULL_VALUE, nil
	}

	n := -1
	for _, b := range barr {
		if n < 0 || len(b) < n {
			n = len(b)
		}
	}

	for i := 0; i < n; i++ {
		cv := value.NewScopeValue(make(map[string]interface{}, len(this.bindings)), item)
		for j, b := range this.bindings {
			cv.SetField(b.Variable(), barr[j][i])
		}

		sv, err := this.satisfies.Evaluate(cv, context)
		if err != nil {
			return nil, err
		}

		if sv.Truth() {
			return value.NewValue(true), nil
		}
	}

	return value.NewValue(false), nil
}

/*
This method returns a pointer to a Formalizer struct
with Allowed set to a new map of type string to interface.
*/
func NewFormalizer() *Formalizer {
	rv := &Formalizer{
		Allowed: value.NewScopeValue(make(map[string]interface{}), nil),
	}

	rv.mapper = rv
	return rv
}

func (this *spanScan) 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

		conn := datastore.NewIndexConnection(context)
		defer notifyConn(conn) // Notify index that I have stopped

		var duration time.Duration
		timer := time.Now()
		defer context.AddPhaseTime("scan", time.Since(timer)-duration)

		go this.scan(context, conn)

		var entry *datastore.IndexEntry
		ok := true
		for ok {
			select {
			case <-this.stopChannel:
				return
			default:
			}

			select {
			case entry, ok = <-conn.EntryChannel():
				t := time.Now()

				if ok {
					cv := value.NewScopeValue(make(map[string]interface{}), parent)
					av := value.NewAnnotatedValue(cv)
					meta := map[string]interface{}{"id": entry.PrimaryKey}
					av.SetAttachment("meta", meta)

					covers := this.plan.Covers()
					if len(covers) > 0 {
						for i, c := range covers {
							if i == 0 {
								av.SetCover(c.Text(), value.NewValue(entry.PrimaryKey))
							} else {
								av.SetCover(c.Text(), entry.EntryKey[i-1])
							}
						}

						av.SetField(this.plan.Term().Alias(), av)
					}

					ok = this.sendItem(av)
				}

				duration += time.Since(t)
			case <-this.stopChannel:
				return
			}
		}
	})
}

func (this *PrimaryScan) scanPrimary(context *Context, parent value.Value) {
	conn := this.newIndexConnection(context)
	defer notifyConn(conn.StopChannel()) // Notify index that I have stopped

	var duration time.Duration
	timer := time.Now()
	defer context.AddPhaseTime("scan", time.Since(timer)-duration)

	go this.scanEntries(context, conn)

	var entry, lastEntry *datastore.IndexEntry

	ok := true
	nitems := 0

	for ok {
		select {
		case <-this.stopChannel:
			return
		default:
		}

		select {
		case entry, ok = <-conn.EntryChannel():
			t := time.Now()

			if ok {
				cv := value.NewScopeValue(make(map[string]interface{}), parent)
				av := value.NewAnnotatedValue(cv)
				av.SetAttachment("meta", map[string]interface{}{"id": entry.PrimaryKey})
				ok = this.sendItem(av)
				lastEntry = entry
				nitems++
			}

			duration += time.Since(t)
		case <-this.stopChannel:
			return
		}

	}

	if conn.Timeout() {
		logging.Errorp("Primary index scan timeout - resorting to chunked scan",
			logging.Pair{"chunkSize", nitems},
			logging.Pair{"startingEntry", lastEntry})
		if lastEntry == nil {
			// no key for chunked scans (primary scan returned 0 items)
			context.Error(errors.NewCbIndexScanTimeoutError(nil))
		}
		// do chunked scans; nitems gives the chunk size, and lastEntry the starting point
		for lastEntry != nil {
			lastEntry = this.scanPrimaryChunk(context, parent, nitems, lastEntry)
		}
	}
}

func (this *DummyScan) 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

		cv := value.NewScopeValue(nil, parent)
		av := value.NewAnnotatedValue(cv)
		this.sendItem(av)
	})
}

/*
Qualify identifiers for the keyspace. It also makes sure that the
keyspace term contains a name or alias.
*/
func (this *KeyspaceRef) Formalize() (f *expression.Formalizer, err error) {
	keyspace := this.Alias()
	if keyspace == "" {
		err = errors.NewNoTermNameError("Keyspace", "plan.keyspace.reference_requires_name_or_alias")
		return
	}

	allowed := value.NewScopeValue(make(map[string]interface{}), nil)
	allowed.SetField(keyspace, keyspace)

	f = expression.NewFormalizer()
	f.Keyspace = keyspace
	f.Allowed = allowed
	return
}

func (this *InitialProject) processTerms(item value.AnnotatedValue, context *Context) bool {
	n := len(this.plan.Terms())
	sv := value.NewScopeValue(make(map[string]interface{}, n), item)
	pv := value.NewAnnotatedValue(sv)
	pv.SetAnnotations(item)

	p := value.NewValue(make(map[string]interface{}, n+32))
	pv.SetAttachment("projection", p)

	for _, term := range this.plan.Terms() {
		if term.Result().Alias() != "" {
			v, err := term.Result().Expression().Evaluate(item, context)
			if err != nil {
				context.Error(errors.NewEvaluationError(err, "projection"))
				return false
			}

			p.SetField(term.Result().Alias(), v)

			// Explicit aliases override data
			if term.Result().As() != "" {
				pv.SetField(term.Result().As(), v)
			}
		} else {
			// Star
			starval := item.GetValue()
			if term.Result().Expression() != nil {
				var err error
				starval, err = term.Result().Expression().Evaluate(item, context)
				if err != nil {
					context.Error(errors.NewEvaluationError(err, "projection"))
					return false
				}
			}

			// Latest star overwrites previous star
			switch sa := starval.Actual().(type) {
			case map[string]interface{}:
				for k, v := range sa {
					p.SetField(k, v)
				}
			}
		}
	}

	return this.sendItem(pv)
}

func (this *InitialProject) processItem(item value.AnnotatedValue, context *Context) bool {
	terms := this.plan.Terms()
	n := len(terms)

	if n > 1 {
		return this.processTerms(item, context)
	}

	if n == 0 {
		return this.sendItem(item)
	}

	// n == 1

	result := terms[0].Result()
	expr := result.Expression()

	if result.Star() && (expr == expression.SELF || expr == nil) {
		// Unprefixed star
		if item.Type() == value.OBJECT {
			return this.sendItem(item)
		} else {
			return this.sendItem(_EMPTY_ANNOTATED_VALUE)
		}
	} else if this.plan.Projection().Raw() {
		// Raw projection of an expression
		v, err := expr.Evaluate(item, context)
		if err != nil {
			context.Error(errors.NewEvaluationError(err, "projection"))
			return false
		}

		if result.As() == "" {
			return this.sendItem(value.NewAnnotatedValue(v))
		}

		sv := value.NewScopeValue(make(map[string]interface{}, 1), item)
		sv.SetField(result.As(), v)
		av := value.NewAnnotatedValue(sv)
		av.SetAttachment("projection", v)
		return this.sendItem(av)
	} else {
		// Any other projection
		return this.processTerms(item, context)
	}
}

func (this *PrimaryScan) scanPrimaryChunk(context *Context, parent value.Value, chunkSize int, indexEntry *datastore.IndexEntry) *datastore.IndexEntry {
	conn, _ := datastore.NewSizedIndexConnection(int64(chunkSize), context)
	conn.SetPrimary()
	defer notifyConn(conn.StopChannel()) // Notify index that I have stopped

	var duration time.Duration
	timer := time.Now()
	defer context.AddPhaseTime("scan", time.Since(timer)-duration)

	go this.scanChunk(context, conn, chunkSize, indexEntry)

	var entry, lastEntry *datastore.IndexEntry

	ok := true
	nitems := 0

	for ok {
		select {
		case <-this.stopChannel:
			return nil
		default:
		}

		select {
		case entry, ok = <-conn.EntryChannel():
			t := time.Now()

			if ok {
				cv := value.NewScopeValue(make(map[string]interface{}), parent)
				av := value.NewAnnotatedValue(cv)
				av.SetAttachment("meta", map[string]interface{}{"id": entry.PrimaryKey})
				ok = this.sendItem(av)
				lastEntry = entry
				nitems++
			}

			duration += time.Since(t)
		case <-this.stopChannel:
			return nil
		}

	}
	logging.Debugp("Primary index chunked scan", logging.Pair{"chunkSize", nitems}, logging.Pair{"lastKey", lastEntry})
	return lastEntry
}

func (this *Let) processItem(item value.AnnotatedValue, context *Context) bool {
	n := len(this.plan.Bindings())
	cv := value.NewScopeValue(make(map[string]interface{}, n), item)
	lv := value.NewAnnotatedValue(cv)
	lv.SetAnnotations(item)

	for _, b := range this.plan.Bindings() {
		v, e := b.Expression().Evaluate(item, context)
		if e != nil {
			context.Error(errors.NewEvaluationError(e, "LET"))
			return false
		}

		lv.SetField(b.Variable(), v)
	}

	return this.sendItem(lv)
}

/*
This method returns a pointer to a Formalizer struct
with allowed set to a new map of type string to interface.
*/
func NewFormalizer(keyspace string, parent *Formalizer) *Formalizer {
	var pv value.Value
	if parent != nil {
		pv = parent.allowed
	}

	rv := &Formalizer{
		keyspace:    keyspace,
		allowed:     value.NewScopeValue(make(map[string]interface{}), pv),
		identifiers: make(map[string]bool),
	}

	if keyspace != "" {
		rv.allowed.SetField(keyspace, keyspace)
	}

	rv.mapper = rv
	return rv
}

/*
Visitor method for Bindings. Value is a new map from string
to interface which is populated using the bindings in the
scope of the parent which is listed by the value allowed.
Bring the bindings that have parrent allowed into scope.
*/
func (this *Formalizer) PushBindings(bindings Bindings) (sv *value.ScopeValue, err error) {
	sv = value.NewScopeValue(make(map[string]interface{}, len(bindings)), this.allowed)

	var expr Expression
	for _, b := range bindings {
		_, ok := this.allowed.Field(b.Variable())
		if ok {
			return nil, fmt.Errorf("Bind alias %s already in scope.", b.Variable())
		}

		expr, err = this.Map(b.Expression())
		if err != nil {
			return nil, err
		}

		b.SetExpression(expr)
		sv.SetField(b.Variable(), b.Variable())
	}

	this.allowed = sv
	return sv, nil
}

func (this *CountScan) 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

		timer := time.Now()

		count, e := this.plan.Keyspace().Count()

		context.AddPhaseTime("count", time.Since(timer))

		if e != nil {
			context.Error(e)
			return
		}

		cv := value.NewScopeValue(nil, parent)
		av := value.NewAnnotatedValue(cv)
		av.SetAttachment("count", value.NewValue(count))
		this.sendItem(av)
	})
}

/*
Fully qualifies the identifiers in the first and second sub-result
using the input parent.
*/
func (this *setOp) Formalize(parent *expression.Formalizer) (*expression.Formalizer, error) {
	_, err := this.first.Formalize(parent)
	if err != nil {
		return nil, err
	}

	_, err = this.second.Formalize(parent)
	if err != nil {
		return nil, err
	}

	terms := this.ResultTerms()
	f := expression.NewFormalizer()
	if parent != nil {
		f.Allowed = value.NewScopeValue(make(map[string]interface{}, len(terms)), parent.Allowed)
	}

	for _, term := range terms {
		f.Allowed.SetField(term.Alias(), term.Alias())
	}

	return f, nil
}

func (this *Array) EvaluateForIndex(item value.Value, context Context) (value.Value, value.Values, error) {
	missing := false
	null := false

	barr := make([][]interface{}, len(this.bindings))
	for i, b := range this.bindings {
		bv, err := b.Expression().Evaluate(item, context)
		if err != nil {
			return nil, nil, err
		}

		if b.Descend() {
			buffer := make([]interface{}, 0, 256)
			bv = value.NewValue(bv.Descendants(buffer))
		}

		switch bv.Type() {
		case value.ARRAY:
			barr[i] = bv.Actual().([]interface{})
		case value.MISSING:
			missing = true
		default:
			null = true
		}
	}

	if missing {
		return value.MISSING_VALUE, nil, nil
	}

	if null {
		return value.NULL_VALUE, nil, nil
	}

	n := -1
	for _, b := range barr {
		if n < 0 || len(b) < n {
			n = len(b)
		}
	}

	var rv []interface{}
	var rvs value.Values
	for i := 0; i < n; i++ {
		cv := value.NewScopeValue(make(map[string]interface{}, len(this.bindings)), item)
		for j, b := range this.bindings {
			cv.SetField(b.Variable(), barr[j][i])
		}

		if this.when != nil {
			wv, e := this.when.Evaluate(cv, context)
			if e != nil {
				return nil, nil, e
			}

			if !wv.Truth() {
				continue
			}
		}

		mv, mvs, e := this.mapping.EvaluateForIndex(cv, context)
		if e != nil {
			return nil, nil, e
		} else if mvs != nil {
			if rvs == nil {
				rvs = make(value.Values, 0, n*8)
			}
			rvs = append(rvs, mvs...)
		} else if mv != nil && mv.Type() != value.MISSING {
			if rv == nil {
				rv = make([]interface{}, 0, n)
			}
			rv = append(rv, mv)
		}
	}

	if rvs != nil {
		return nil, rvs, nil
	} else {
		if rv == nil {
			rv = make([]interface{}, 0)
		}
		return value.NewValue(rv), nil, nil
	}
}