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 *ValueScan) 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

		pairs := this.plan.Values()
		for _, pair := range pairs {
			key, err := pair.Key.Evaluate(parent, context)
			if err != nil {
				context.Error(errors.NewEvaluationError(err, "VALUES"))
				return
			}

			val, err := pair.Value.Evaluate(parent, context)
			if err != nil {
				context.Error(errors.NewEvaluationError(err, "VALUES"))
				return
			}

			av := value.NewAnnotatedValue(nil)
			av.SetAttachment("key", key)
			av.SetAttachment("value", val)

			if !this.sendItem(av) {
				return
			}
		}
	})
}

func (b *activeRequestsKeyspace) Fetch(keys []string) ([]datastore.AnnotatedPair, []errors.Error) {
	var errs []errors.Error
	rv := make([]datastore.AnnotatedPair, 0, len(keys))

	server.ActiveRequestsForEach(func(id string, request server.Request) {
		item := value.NewAnnotatedValue(map[string]interface{}{
			"RequestId":     id,
			"RequestTime":   request.RequestTime().String(),
			"ElapsedTime":   time.Since(request.RequestTime()).String(),
			"ExecutionTime": time.Since(request.ServiceTime()).String(),
			"State":         request.State(),
		})
		if request.Statement() != "" {
			item.SetField("Statement", request.Statement())
		}
		if request.Prepared() != nil {
			p := request.Prepared()
			item.SetField("Prepared.Name", p.Name())
			item.SetField("Prepared.Text", p.Text())
		}
		item.SetAttachment("meta", map[string]interface{}{
			"id": id,
		})
		rv = append(rv, datastore.AnnotatedPair{
			Key:   id,
			Value: item,
		})
	})
	return rv, errs
}

func (b *requestLogKeyspace) Fetch(keys []string) ([]datastore.AnnotatedPair, []errors.Error) {
	var errs []errors.Error
	rv := make([]datastore.AnnotatedPair, 0, len(keys))

	accounting.RequestsForeach(func(id string, entry *accounting.RequestLogEntry) {
		item := value.NewAnnotatedValue(map[string]interface{}{
			"RequestId":   id,
			"ElapsedTime": entry.ElapsedTime,
			"ServiceTime": entry.ServiceTime,
			"ResultCount": entry.ResultCount,
			"ResultSize":  entry.ResultSize,
			"ErrorCount":  entry.ErrorCount,
			"SortCount":   entry.SortCount,
			"Time":        entry.Time.String(),
		})
		if entry.Statement != "" {
			item.SetField("Statement", entry.Statement)
		}
		if entry.PreparedName != "" {
			item.SetField("PreparedName", entry.PreparedName)
			item.SetField("PreparedText", entry.PreparedText)
		}
		item.SetAttachment("meta", map[string]interface{}{
			"id": id,
		})
		rv = append(rv, datastore.AnnotatedPair{
			Key:   id,
			Value: item,
		})
	})
	return rv, errs
}

func (this *IndexNest) fetch(entries []*datastore.IndexEntry, context *Context) (
	[]interface{}, bool) {
	// Build list of keys
	var keys []string
	if len(entries) <= 16 {
		keys = _NEST_INDEX_STRING_POOL.Get()
		defer _NEST_INDEX_STRING_POOL.Put(keys)
	} else {
		keys = make([]string, 0, len(entries))
	}

	for _, e := range entries {
		keys = append(keys, e.PrimaryKey)
	}

	// Fetch
	pairs, errs := this.plan.Keyspace().Fetch(keys)

	fetchOk := true
	for _, err := range errs {
		context.Error(err)
		if err.IsFatal() {
			fetchOk = false
		}
	}

	if len(pairs) == 0 {
		return nil, fetchOk
	}

	projection := this.plan.Term().Projection()
	nvs := make([]interface{}, 0, len(pairs))
	for _, pair := range pairs {
		nestItem := pair.Value
		var nv value.AnnotatedValue

		// Apply projection, if any
		if projection != nil {
			projectedItem, e := projection.Evaluate(nestItem, context)
			if e != nil {
				context.Error(errors.NewEvaluationError(e, "nest path"))
				return nil, false
			}

			if projectedItem.Type() == value.MISSING {
				continue
			}

			nv = value.NewAnnotatedValue(projectedItem)
			nv.SetAnnotations(nestItem)
		} else {
			nv = nestItem
		}

		nvs = append(nvs, nv)
	}

	return nvs, fetchOk
}

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 *FinalProject) processItem(item value.AnnotatedValue, context *Context) bool {
	pv := item.GetAttachment("projection")
	if pv != nil {
		v := pv.(value.Value)
		return this.sendItem(value.NewAnnotatedValue(v))
	}

	return this.sendItem(item)
}

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 *Prepare) 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
		value := value.NewAnnotatedValue(this.plan)
		this.sendItem(value)

	})
}

// generate a mock document - used by fetchOne to mock a document in the keyspace
func genItem(i int, nitems int) (value.AnnotatedValue, errors.Error) {
	if i < 0 || i >= nitems {
		return nil, errors.NewOtherDatastoreError(nil,
			fmt.Sprintf("item out of mock range: %v [0,%v)", i, nitems))
	}
	id := strconv.Itoa(i)
	doc := value.NewAnnotatedValue(map[string]interface{}{"id": id, "i": float64(i)})
	doc.SetAttachment("meta", map[string]interface{}{"id": id})
	return doc, nil
}

func (b *storeKeyspace) fetchOne(key string) (value.AnnotatedValue, errors.Error) {
	if key == b.namespace.store.actualStore.Id() {
		doc := value.NewAnnotatedValue(map[string]interface{}{
			"id":  b.namespace.store.actualStore.Id(),
			"url": b.namespace.store.actualStore.URL(),
		})
		return doc, nil
	}
	return nil, errors.NewSystemDatastoreError(nil, "Key Not Found "+key)
}

func (this *Unnest) processItem(item value.AnnotatedValue, context *Context) bool {
	ev, err := this.plan.Term().Expression().Evaluate(item, context)
	if err != nil {
		context.Error(errors.NewEvaluationError(err, "UNNEST path"))
		return false
	}

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

	acts := actuals.([]interface{})
	if len(acts) == 0 {
		// Outer unnest
		return !this.plan.Term().Outer() || this.sendItem(item)
	}

	// Attach and send
	for i, act := range acts {
		var av value.AnnotatedValue
		if i < len(acts)-1 {
			av = value.NewAnnotatedValue(item.Copy())
		} else {
			av = item
		}

		actv := value.NewAnnotatedValue(act)
		actv.SetAttachment("unnest_position", i)
		av.SetField(this.plan.Alias(), actv)

		if !this.sendItem(av) {
			return false
		}
	}

	return true
}

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

func (b *namespaceKeyspace) fetchOne(key string) (value.AnnotatedValue, errors.Error) {
	namespace, excp := b.namespace.store.actualStore.NamespaceById(key)
	if namespace != nil {
		doc := value.NewAnnotatedValue(map[string]interface{}{
			"id":           namespace.Id(),
			"name":         namespace.Name(),
			"datastore_id": b.namespace.store.actualStore.Id(),
		})
		return doc, nil
	}
	return nil, errors.NewSystemDatastoreError(excp, "Key Not Found "+key)
}

func fetch(path string) (item value.AnnotatedValue, e errors.Error) {
	bytes, er := ioutil.ReadFile(path)
	if er != nil {
		return nil, errors.NewFileDatastoreError(er, "")
	}

	doc := value.NewAnnotatedValue(value.NewValue(bytes))
	doc.SetAttachment("meta", map[string]interface{}{"id": documentPathToId(path)})
	item = doc

	return
}

// N1QLTransform will use compiled list of expression from N1QL's DDL
// statement and evaluate a document using them to return a secondary
// key as JSON object.
// `meta` supplies a dictionary of,
//      `id`, `byseqno`, `revseqno`, `flags`, `expiry`, `locktime`,
//      `nru`
func N1QLTransform(
	docid, doc []byte, cExprs []interface{},
	meta map[string]interface{}) ([]byte, error) {

	arrValue := make([]qvalue.Value, 0, len(cExprs))
	context := qexpr.NewIndexContext()
	skip := true
	docval := qvalue.NewAnnotatedValue(doc)
	docval.SetAttachment("meta", meta)
	for _, cExpr := range cExprs {
		expr := cExpr.(qexpr.Expression)
		key, err := expr.Evaluate(docval, context)
		if err != nil {
			return nil, err

		} else if key.Type() == qvalue.MISSING && skip {
			return nil, nil

		} else if key.Type() == qvalue.MISSING {
			arrValue = append(arrValue, missing)
			continue
		}
		skip = false
		arrValue = append(arrValue, key)
	}

	if len(cExprs) == 1 && len(arrValue) == 1 && docid == nil {
		// used for partition-key evaluation and where predicate.
		// Marshal partition-key and where as a basic JSON data-type.
		return arrValue[0].MarshalJSON()

	} else if len(arrValue) > 0 {
		// The shape of the secondary key looks like,
		//     [expr1] - for simple key
		//     [expr1, expr2] - for composite key

		// in case we need to append docid to skeys, it is applicable
		// only when docid is not `nil`
		//if docid != nil {
		//    arrValue = append(arrValue, qvalue.NewValue(string(docid)))
		//}
		secKey := qvalue.NewValue(make([]interface{}, len(arrValue)))
		for i, key := range arrValue {
			secKey.SetIndex(i, key)
		}
		return secKey.MarshalJSON() // return as JSON array
	}
	return nil, nil
}

func (b *keyspace) Fetch(keys []string) ([]datastore.AnnotatedPair, []errors.Error) {

	if len(keys) == 0 {
		return nil, nil
	}

	bulkResponse, err := b.cbbucket.GetBulk(keys)
	if err != nil {
		// Ignore "Not found" keys
		if !isNotFoundError(err) {
			return nil, []errors.Error{errors.NewCbBulkGetError(err, "")}
		}
	}

	i := 0
	rv := make([]datastore.AnnotatedPair, len(bulkResponse))
	for k, v := range bulkResponse {

		var doc datastore.AnnotatedPair
		doc.Key = k

		Value := value.NewAnnotatedValue(value.NewValue(v.Body))

		meta_flags := binary.BigEndian.Uint32(v.Extras[0:4])
		meta_type := "json"
		if Value.Type() == value.BINARY {
			meta_type = "base64"
		}
		Value.SetAttachment("meta", map[string]interface{}{
			"id":    k,
			"cas":   v.Cas,
			"type":  meta_type,
			"flags": uint32(meta_flags),
		})

		// Uncomment when needed
		//logging.Debugf("CAS Value for key %v is %v flags %v", k, uint64(v.Cas), meta_flags)

		doc.Value = Value
		rv[i] = doc
		i++

	}

	logging.Debugf("Fetched %d keys ", i)

	return rv, nil
}

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

/*
Add input item to the cumulative set. Get the set. If
no errors enountered add the item to the set and return
it. If set has not been initialized yet, create a new set
with capacity _OBJECT_CAP and add the item. Return the
set value.
*/
func setAdd(item, cumulative value.Value) (value.AnnotatedValue, error) {
	av, ok := cumulative.(value.AnnotatedValue)
	if !ok {
		av = value.NewAnnotatedValue(cumulative)
	}

	set, e := getSet(av)
	if e == nil {
		set.Add(item)
		return av, nil
	}

	set = value.NewSet(_OBJECT_CAP)
	set.Add(item)
	av.SetAttachment("set", set)
	return av, nil
}