func (this *IntersectScan) processKey(item value.AnnotatedValue, context *Context) bool {
	m := item.GetAttachment("meta")
	meta, ok := m.(map[string]interface{})
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Missing or invalid meta %v of type %T.", m, m)))
		return false
	}

	k := meta["id"]
	key, ok := k.(string)
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Missing or invalid primary key %v of type %T.", k, k)))
		return false
	}

	count := this.counts[key]
	this.counts[key] = count + 1

	if count+1 == len(this.scans) {
		delete(this.values, key)
		return this.sendItem(item)
	}

	if count == 0 {
		this.values[key] = item
	}

	return true
}

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 *DummyFetch) processItem(item value.AnnotatedValue, context *Context) bool {
	item.SetField(this.plan.Term().Alias(), item)
	if !this.sendItem(item) {
		return false
	}

	return true
}

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 *Distinct) processItem(item value.AnnotatedValue, context *Context) bool {
	p := item.GetAttachment("projection")
	if p == nil {
		p = item
	}

	this.set.Put(p.(value.Value), item)
	return true
}

func (this *Collect) processItem(item value.AnnotatedValue, context *Context) bool {
	if len(this.values) == cap(this.values) {
		values := make([]interface{}, len(this.values), len(this.values)<<1)
		copy(values, this.values)
		this.values = values
	}

	this.values = append(this.values, item.Actual())
	return true
}

func (this *Merge) processMatch(item value.AnnotatedValue,
	context *Context, update, delete, insert Operator) bool {
	kv, e := this.plan.Key().Evaluate(item, context)
	if e != nil {
		context.Error(errors.NewEvaluationError(e, "MERGE key"))
		return false
	}

	ka := kv.Actual()
	k, ok := ka.(string)
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Invalid MERGE key %v of type %T.", ka, ka)))
		return false
	}

	timer := time.Now()

	fetchOk := true
	bvs, errs := this.plan.Keyspace().Fetch([]string{k})

	this.duration += time.Since(timer)

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

	if len(bvs) > 0 {
		bv := bvs[0]
		item.SetField(this.plan.KeyspaceRef().Alias(), bv.Value)

		// Perform UPDATE and/or DELETE
		if update != nil {
			update.Input().ItemChannel() <- item
		}

		if delete != nil {
			delete.Input().ItemChannel() <- item
		}
	} else {
		// Not matched; INSERT
		if insert != nil {
			insert.Input().ItemChannel() <- item
		}
	}

	return fetchOk
}

func (this *Unset) processItem(item value.AnnotatedValue, context *Context) bool {
	clone, ok := item.GetAttachment("clone").(value.AnnotatedValue)
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Invalid UPDATE clone of type %T.", clone)))
		return false
	}

	for _, t := range this.plan.Node().Terms() {
		unsetPath(t, clone, context)
	}

	return this.sendItem(item)
}

func (this *IntermediateGroup) processItem(item value.AnnotatedValue, context *Context) bool {
	// Generate the group key
	var gk string
	if len(this.plan.Keys()) > 0 {
		var e error
		gk, e = groupKey(item, this.plan.Keys(), context)
		if e != nil {
			context.Fatal(errors.NewEvaluationError(e, "GROUP key"))
			return false
		}
	}

	// Get or seed the group value
	gv := this.groups[gk]
	if gv == nil {
		gv = item
		this.groups[gk] = gv
		return true
	}

	// Cumulate aggregates
	part, ok := item.GetAttachment("aggregates").(map[string]value.Value)
	if !ok {
		context.Fatal(errors.NewInvalidValueError(
			fmt.Sprintf("Invalid partial aggregates %v of type %T", part, part)))
		return false
	}

	cumulative := gv.GetAttachment("aggregates").(map[string]value.Value)
	if !ok {
		context.Fatal(errors.NewInvalidValueError(
			fmt.Sprintf("Invalid cumulative aggregates %v of type %T", cumulative, cumulative)))
		return false
	}

	for _, agg := range this.plan.Aggregates() {
		a := agg.String()
		v, e := agg.CumulateIntermediate(part[a], cumulative[a], context)
		if e != nil {
			context.Fatal(errors.NewGroupUpdateError(
				e, "Error updating intermediate GROUP value."))
			return false
		}

		cumulative[a] = v
	}

	return true
}

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 *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 *IndexNest) nestEntries(item value.AnnotatedValue,
	entries []*datastore.IndexEntry, context *Context) (ok bool) {
	var nvs []interface{}

	if len(entries) > 0 {
		covers := this.plan.Covers()
		if len(covers) == 0 {
			nvs, ok = this.fetch(entries, context)
			if !ok {
				return ok
			}
		} else {
			nvs = make([]interface{}, len(entries))
			for i, entry := range entries {
				nv := value.NewAnnotatedValue(nil)
				meta := map[string]interface{}{"id": entry.PrimaryKey}
				nv.SetAttachment("meta", meta)

				for i, c := range covers {
					nv.SetCover(c.Text(), entry.EntryKey[i])
				}

				nvs[i] = nv
			}
		}
	}

	if len(nvs) > 0 {
		item.SetField(this.plan.Term().Alias(), nvs)
	} else {
		if !this.plan.Outer() {
			return true
		}

		item.SetField(this.plan.Term().Alias(), value.EMPTY_ARRAY_VALUE)
	}

	return this.sendItem(item)
}

func (this *Order) lessThan(v1 value.AnnotatedValue, v2 value.AnnotatedValue) bool {
	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
}

func (this *base) requireKey(item value.AnnotatedValue, context *Context) (string, bool) {
	mv := item.GetAttachment("meta")
	if mv == nil {
		context.Error(errors.NewError(nil, "Unable to find meta."))
		return "", false
	}

	meta := mv.(map[string]interface{})
	key, ok := meta["id"]
	if !ok {
		context.Error(errors.NewError(nil, "Unable to find key."))
		return "", false
	}

	act := value.NewValue(key).Actual()
	switch act := act.(type) {
	case string:
		return act, true
	default:
		e := errors.NewError(nil, fmt.Sprintf("Unable to process non-string key %v of type %T.", act, act))
		context.Error(e)
		return "", false
	}
}

func (this *UnionScan) processKey(item value.AnnotatedValue, context *Context) bool {
	m := item.GetAttachment("meta")
	meta, ok := m.(map[string]interface{})
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Missing or invalid meta %v of type %T.", m, m)))
		return false
	}

	k := meta["id"]
	key, ok := k.(string)
	if !ok {
		context.Error(errors.NewInvalidValueError(
			fmt.Sprintf("Missing or invalid primary key %v of type %T.", k, k)))
		return false
	}

	if this.values[key] != nil {
		return true
	}

	this.values[key] = item
	return this.sendItem(item)
}

func (this *IndexJoin) joinEntry(item value.AnnotatedValue,
	entry *datastore.IndexEntry, context *Context) (found, ok bool) {
	var jv value.AnnotatedValue
	covers := this.plan.Covers()

	if len(covers) == 0 {
		jv, ok = this.fetch(entry, context)
		if jv == nil || !ok {
			return jv != nil, ok
		}
	} else {
		jv = value.NewAnnotatedValue(nil)
		meta := map[string]interface{}{"id": entry.PrimaryKey}
		jv.SetAttachment("meta", meta)

		for i, c := range covers {
			jv.SetCover(c.Text(), entry.EntryKey[i])
		}
	}

	joined := item.Copy().(value.AnnotatedValue)
	joined.SetField(this.plan.Term().Alias(), jv)
	return true, this.sendItem(joined)
}

func (this *Join) processItem(item value.AnnotatedValue, context *Context) bool {
	kv, e := this.plan.Term().Keys().Evaluate(item, context)
	if e != nil {
		context.Error(errors.NewEvaluationError(e, "JOIN keys"))
		return false
	}

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

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

	// Build list of keys
	keys := make([]string, 0, len(acts))
	for _, key := range acts {
		k := value.NewValue(key).Actual()
		switch k := k.(type) {
		case string:
			keys = append(keys, k)
		}
	}

	timer := time.Now()

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

	this.duration += time.Since(timer)

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

	found := len(pairs) > 0

	// Attach and send
	for i, pair := range pairs {
		joinItem := pair.Value
		var jv value.AnnotatedValue

		// Apply projection, if any
		projection := this.plan.Term().Projection()
		if projection != nil {
			projectedItem, e := projection.Evaluate(joinItem, context)
			if e != nil {
				context.Error(errors.NewEvaluationError(e, "join path"))
				return false
			}

			jv = value.NewAnnotatedValue(projectedItem)
			jv.SetAnnotations(joinItem)
		} else {
			jv = joinItem
		}

		var av value.AnnotatedValue
		if i < len(pairs)-1 {
			av = value.NewAnnotatedValue(item.Copy())
		} else {
			av = item
		}

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

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

	return (found || !this.plan.Outer() || this.sendItem(item)) && fetchOk
}

func (this *Clone) processItem(item value.AnnotatedValue, context *Context) bool {
	clone := item.CopyForUpdate()
	item.SetAttachment("clone", clone)
	return this.sendItem(item)
}

func (this *Fetch) flushBatch(context *Context) bool {
	defer this.releaseBatch()

	if len(this.batch) == 0 {
		return true
	}

	keys := _STRING_POOL.Get()
	defer _STRING_POOL.Put(keys)

	batchMap := _STRING_ANNOTATED_POOL.Get()
	defer _STRING_ANNOTATED_POOL.Put(batchMap)

	for _, av := range this.batch {
		meta := av.GetAttachment("meta")

		switch meta := meta.(type) {
		case map[string]interface{}:
			key := meta["id"]
			act := value.NewValue(key).Actual()
			switch act := act.(type) {
			case string:
				keys = append(keys, act)
				batchMap[act] = av
			default:
				context.Error(errors.NewInvalidValueError(fmt.Sprintf(
					"Missing or invalid primary key %v of type %T.",
					act, act)))
				return false
			}
		default:
			context.Error(errors.NewInvalidValueError(
				"Missing or invalid meta for primary key."))
			return false
		}
	}

	timer := time.Now()

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

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

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

	fetchMap := _STRING_ANNOTATED_POOL.Get()
	defer _STRING_ANNOTATED_POOL.Put(fetchMap)

	// Attach meta
	for _, pair := range pairs {
		pv, ok := pair.Value.(value.AnnotatedValue)
		if !ok {
			context.Fatal(errors.NewInvalidValueError(fmt.Sprintf(
				"Invalid fetch value %v of type %T", pair.Value)))
			return false
		}

		var fv value.AnnotatedValue

		// Apply projection, if any
		projection := this.plan.Term().Projection()
		if projection != nil {
			projectedItem, e := projection.Evaluate(pv, context)
			if e != nil {
				context.Error(errors.NewEvaluationError(e, "fetch path"))
				return false
			}

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

			fv = value.NewAnnotatedValue(projectedItem)
			fv.SetAnnotations(pv)
		} else {
			fv = pv
		}

		fetchMap[pair.Key] = fv
	}

	// Preserve order of keys
	for _, key := range keys {
		fv := fetchMap[key]
		if fv == nil {
			continue
		}

		item := batchMap[key]
		item.SetField(this.plan.Term().Alias(), fv)

		if !this.sendItem(item) {
			return false
//.........这里部分代码省略.........