/*
Fully qualify identifiers for each of the constituent fields
in the merge source statement.
*/
func (this *MergeSource) Formalize() (f *expression.Formalizer, err error) {
	if this.from != nil {
		_, err = this.from.Formalize(expression.NewFormalizer())
		if err != nil {
			return
		}
	}

	if this.query != nil {
		err = this.query.Formalize()
		if err != nil {
			return
		}
	}

	keyspace := this.Alias()
	if keyspace == "" {
		return nil, fmt.Errorf("MergeSource missing alias.")
	}

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

/*
Fully qualify identifiers for each of the constituent clauses
in the merge statement.
*/
func (this *Merge) Formalize() (err error) {
	kf, err := this.keyspace.Formalize()
	if err != nil {
		return err
	}

	sf, err := this.source.Formalize()
	if err != nil {
		return err
	}

	this.key, err = sf.Map(this.key)
	if err != nil {
		return err
	}

	if kf.Keyspace() != "" &&
		kf.Keyspace() == sf.Keyspace() {
		return fmt.Errorf("Duplicate alias %s.", kf.Keyspace())
	}

	f := expression.NewFormalizer("", nil)

	if kf.Keyspace() != "" {
		f.Allowed().SetField(kf.Keyspace(), kf.Keyspace())
	}

	if sf.Keyspace() != "" {
		f.Allowed().SetField(sf.Keyspace(), sf.Keyspace())
	}

	err = this.actions.MapExpressions(f)
	if err != nil {
		return
	}

	if this.limit != nil {
		_, err = this.limit.Accept(expression.NewFormalizer("", nil))
		if err != nil {
			return
		}
	}

	if this.returning != nil {
		_, err = this.returning.Formalize(kf)
	}

	return
}

/*
Fully qualify identifiers for each of the constituent clauses
in the delete statement.
*/
func (this *Delete) Formalize() (err error) {
	f, err := this.keyspace.Formalize()
	if err != nil {
		return err
	}

	empty := expression.NewFormalizer("", nil)
	if this.keys != nil {
		_, err = this.keys.Accept(empty)
		if err != nil {
			return
		}
	}

	if this.where != nil {
		this.where, err = f.Map(this.where)
		if err != nil {
			return
		}
	}

	if this.limit != nil {
		_, err = this.limit.Accept(empty)
		if err != nil {
			return
		}
	}

	if this.returning != nil {
		_, err = this.returning.Formalize(f)
	}

	return
}

/*
This method fully qualifies the identifiers for each term
in the result expression. It disallows duplicate alias and
exempts explicit aliases from being formalized.
*/
func (this *Projection) Formalize(in *expression.Formalizer) (f *expression.Formalizer, err error) {
	// Disallow duplicate aliases
	aliases := make(map[string]bool, len(this.terms))
	for _, term := range this.terms {
		if term.alias == "" {
			continue
		}

		if aliases[term.alias] {
			return nil, fmt.Errorf("Duplicate result alias %s.", term.alias)
		}

		aliases[term.alias] = true
	}

	f = expression.NewFormalizer()
	f.Allowed = in.Allowed.Copy()
	f.Keyspace = in.Keyspace

	err = this.MapExpressions(f)
	if err != nil {
		return
	}

	// Exempt explicit aliases from being formalized
	for _, term := range this.terms {
		if term.as != "" {
			f.Allowed.SetField(term.as, term.as)
		}
	}

	return
}

/*
Fully qualify identifiers for each of the constituent clauses
in the upsert statement.
*/
func (this *Upsert) Formalize() (err error) {
	if this.values != nil {
		f := expression.NewFormalizer("", nil)
		err = this.values.MapExpressions(f)
		if err != nil {
			return
		}
	}

	if this.query != nil {
		err = this.query.Formalize()
		if err != nil {
			return
		}
	}

	f, err := this.keyspace.Formalize()
	if err != nil {
		return err
	}

	if this.returning != nil {
		_, err = this.returning.Formalize(f)
	}

	return
}

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

/*
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 (this *SelectTerm) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	err = this.query.FormalizeSubquery(parent)
	if err != nil {
		return nil, err
	}

	return expression.NewFormalizer("", nil), nil
}

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

	f = expression.NewFormalizer(keyspace, nil)
	return
}

func (this *builder) buildScan(keyspace datastore.Keyspace, node *algebra.KeyspaceTerm, limit expression.Expression) (
	secondary plan.Operator, primary *plan.PrimaryScan, err error) {
	var indexes, hintIndexes, otherIndexes []datastore.Index
	hints := node.Indexes()
	if hints != nil {
		indexes, err = allHints(keyspace, hints)
		hintIndexes = indexes
	} else {
		indexes, err = allIndexes(keyspace)
		otherIndexes = indexes
	}

	if err != nil {
		return
	}

	pred := this.where
	if pred != nil {
		dnf := NewDNF()
		pred = pred.Copy()
		pred, err = dnf.Map(pred)
		if err != nil {
			return
		}

		formalizer := expression.NewFormalizer()
		formalizer.Keyspace = node.Alias()
		primaryKey := expression.Expressions{
			expression.NewField(
				expression.NewMeta(expression.NewConstant(node.Alias())),
				expression.NewFieldName("id", false)),
		}

		sargables, er := sargableIndexes(indexes, pred, primaryKey, dnf, formalizer)
		if er != nil {
			return nil, nil, er
		}

		minimals, er := minimalIndexes(sargables, pred)
		if er != nil {
			return nil, nil, er
		}

		if len(minimals) > 0 {
			secondary, err = this.buildSecondaryScan(minimals, node, limit)
			return secondary, nil, err
		}
	}

	primary, err = this.buildPrimaryScan(keyspace, node, limit, hintIndexes, otherIndexes)
	return nil, primary, err
}

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

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

	return expression.NewFormalizer(), nil
}

/*
This method qualifies identifiers for all the contituent
clauses namely the from, let, where, group and projection
in a subselect statement.It calls Formalize for the from,
group and projections, calls Map to map the where
expressions and calls PushBindings for the let clause.
*/
func (this *Subselect) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	if this.from != nil {
		f, err = this.from.Formalize(parent)
		if err != nil {
			return nil, err
		}
	} else {
		f = expression.NewFormalizer("", parent)
	}

	if this.let != nil {
		_, err = f.PushBindings(this.let)
		if err != nil {
			return nil, err
		}
	}

	if this.where != nil {
		this.where, err = f.Map(this.where)
		if err != nil {
			return nil, err
		}
	}

	if this.group != nil {
		err = this.group.Formalize(f)
		if err != nil {
			return nil, err
		}
	}

	f, err = this.projection.Formalize(f)
	if err != nil {
		return nil, err
	}

	// Determine if this is a correlated subquery
	this.correlated = false
	immediate := f.Allowed().GetValue().Fields()

	for ident, _ := range f.Identifiers() {
		if _, ok := immediate[ident]; !ok {
			this.correlated = true
			break
		}
	}

	return f, nil
}

/*
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 *builder) buildJoinScan(keyspace datastore.Keyspace, node *algebra.KeyspaceTerm, op string) (
	datastore.Index, expression.Covers, error) {
	indexes, err := allIndexes(keyspace)
	if err != nil {
		return nil, nil, err
	}

	var pred expression.Expression
	pred = expression.NewIsNotNull(node.Keys().Copy())
	dnf := NewDNF()
	pred, err = dnf.Map(pred)
	if err != nil {
		return nil, nil, err
	}

	subset := pred
	if this.where != nil {
		subset = expression.NewAnd(subset, this.where.Copy())
		subset, err = dnf.Map(subset)
		if err != nil {
			return nil, nil, err
		}
	}

	formalizer := expression.NewFormalizer()
	formalizer.Keyspace = node.Alias()
	primaryKey := expression.Expressions{
		expression.NewField(
			expression.NewMeta(expression.NewConstant(node.Alias())),
			expression.NewFieldName("id", false)),
	}

	sargables, err := sargableIndexes(indexes, pred, subset, primaryKey, dnf, formalizer)
	if err != nil {
		return nil, nil, err
	}

	minimals, err := minimalIndexes(sargables, pred)
	if err != nil {
		return nil, nil, err
	}

	if len(minimals) == 0 {
		return nil, nil, errors.NewNoIndexJoinError(node.Alias(), op)
	}

	return this.buildCoveringJoinScan(minimals, node, op)
}

/*
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("", parent)
	for _, term := range terms {
		f.Allowed().SetField(term.Alias(), term.Alias())
	}

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

	f = expression.NewFormalizer(alias, parent)
	return
}

/*
Qualify all identifiers for the parent expression. Checks is
a nest alias exists and if it is a duplicate alias.
*/
func (this *IndexNest) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	f, err = this.left.Formalize(parent)
	if err != nil {
		return
	}

	_, ok := f.Allowed.Field(this.keyFor)
	if !ok {
		err = errors.NewUnknownForError("NEST", this.keyFor, "plan.nest.unknown_for")
		return nil, err
	}

	alias := this.Alias()
	if alias == "" {
		err = errors.NewNoTermNameError("NEST", "plan.nest.requires_name_or_alias")
		return nil, err
	}

	_, ok = f.Allowed.Field(alias)
	if ok {
		err = errors.NewDuplicateAliasError("NEST", alias, "plan.nest.duplicate_alias")
		return nil, err
	}

	f.Allowed.SetField(alias, alias)
	f.Keyspace = ""

	var p *expression.Formalizer
	if parent != nil {
		p = parent.Copy()
	} else {
		p = expression.NewFormalizer()
	}

	p.Allowed.SetField(alias, alias)
	this.right.keys, err = p.Map(this.right.keys)
	return
}

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

/*
Qualify all identifiers for the parent expression. Checks is
a nest alias exists and if it is a duplicate alias.
*/
func (this *IndexNest) Formalize(parent *expression.Formalizer) (f *expression.Formalizer, err error) {
	f, err = this.left.Formalize(parent)
	if err != nil {
		return
	}

	_, ok := f.Allowed().Field(this.keyFor)
	if !ok {
		err = errors.NewUnknownForError("NEST", this.keyFor, "plan.nest.unknown_for")
		return nil, err
	}

	alias := this.Alias()
	if alias == "" {
		err = errors.NewNoTermNameError("NEST", "plan.nest.requires_name_or_alias")
		return nil, err
	}

	_, ok = f.Allowed().Field(alias)
	if ok {
		err = errors.NewDuplicateAliasError("NEST", alias, "plan.nest.duplicate_alias")
		return nil, err
	}

	f.Allowed().SetField(alias, alias)
	f.SetKeyspace("")

	p := expression.NewFormalizer("", parent)
	p.Allowed().SetField(alias, alias)
	this.right.keys, err = p.Map(this.right.keys)

	for ident, val := range p.Identifiers() {
		f.Identifiers()[ident] = val
	}

	return
}

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

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

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

	f.SetKeyspace(keyspace)
	return
}