func (this *builder) selectScan(keyspace datastore.Keyspace, node *algebra.KeyspaceTerm,
	limit expression.Expression) (op plan.Operator, err error) {
	keys := node.Keys()
	if keys != nil {
		switch keys := keys.(type) {
		case *expression.ArrayConstruct:
			this.maxParallelism = util.MaxInt(1, len(keys.Operands()))
		case *algebra.NamedParameter, *algebra.PositionalParameter:
			this.maxParallelism = 0
		default:
			this.maxParallelism = 1
		}

		return plan.NewKeyScan(keys), nil
	}

	this.maxParallelism = 0 // Use default parallelism for index scans

	secondary, primary, err := this.buildScan(keyspace, node, limit)
	if err != nil {
		return nil, err
	}

	if secondary != nil {
		return secondary, nil
	} else {
		return primary, nil
	}
}

func (this *builder) buildCoveringJoinScan(secondaries map[datastore.Index]*indexEntry,
	node *algebra.KeyspaceTerm, op string) (datastore.Index, expression.Covers, error) {
	if this.cover == nil {
		for index, _ := range secondaries {
			return index, nil, nil
		}
	}

	alias := node.Alias()
	exprs := this.cover.Expressions()

outer:
	for index, entry := range secondaries {
		for _, expr := range exprs {
			if !expr.CoveredBy(alias, entry.keys) {
				continue outer
			}
		}

		covers := make(expression.Covers, len(entry.keys))
		for i, key := range entry.keys {
			covers[i] = expression.NewCover(key)
		}

		return index, covers, nil
	}

	for index, _ := range secondaries {
		return index, nil, nil
	}

	return nil, nil, errors.NewNoIndexJoinError(node.Alias(), op)
}

func (this *builder) buildCoveringScan(secondaries map[datastore.Index]*indexEntry,
	node *algebra.KeyspaceTerm, limit expression.Expression) (plan.Operator, error) {
	if this.cover == nil {
		return nil, nil
	}

	alias := node.Alias()
	exprs := this.cover.Expressions()
	id := expression.NewField(
		expression.NewMeta(expression.NewIdentifier(node.Alias())),
		expression.NewFieldName("id", false))

outer:
	for index, entry := range secondaries {
		keys := entry.keys
		if !index.IsPrimary() {
			// Matches execution.spanScan.RunOnce()
			keys = append(keys, id)
		}

		// Use the first available covering index
		for _, expr := range exprs {
			if !expr.CoveredBy(alias, keys) {
				continue outer
			}
		}

		covers := make(expression.Covers, 0, len(keys))
		for _, key := range keys {
			covers = append(covers, expression.NewCover(key))
		}

		scan := plan.NewIndexScan(index, node, entry.spans, false, limit, covers)
		this.coveringScan = scan

		if len(entry.spans) > 1 {
			// Use UnionScan to de-dup multiple spans
			return plan.NewUnionScan(scan), nil
		}

		return scan, nil
	}

	return nil, nil
}

func (this *builder) VisitKeyspaceTerm(node *algebra.KeyspaceTerm) (interface{}, error) {
	node.SetDefaultNamespace(this.namespace)
	keyspace, err := this.getTermKeyspace(node)
	if err != nil {
		return nil, err
	}

	if this.subquery && this.correlated && node.Keys() == nil {
		return nil, errors.NewSubqueryMissingKeysError(node.Keyspace())
	}

	scan, err := this.selectScan(keyspace, node, this.limit)
	if err != nil {
		return nil, err
	}

	this.children = append(this.children, scan)

	if this.coveringScan == nil {
		fetch := plan.NewFetch(keyspace, node)
		this.subChildren = append(this.subChildren, fetch)
	}

	return nil, nil
}

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

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
}

func (this *builder) getTermKeyspace(node *algebra.KeyspaceTerm) (datastore.Keyspace, error) {
	node.SetDefaultNamespace(this.namespace)
	ns := node.Namespace()

	datastore := this.datastore
	if strings.ToLower(ns) == "#system" {
		datastore = this.systemstore
	}

	namespace, err := datastore.NamespaceByName(ns)
	if err != nil {
		return nil, err
	}

	keyspace, err := namespace.KeyspaceByName(node.Keyspace())
	if err != nil {
		return nil, err
	}

	return keyspace, nil
}