func (this *CollectionFirstOperator) Evaluate(item *dparval.Value) (*dparval.Value, error) {
	// first evaluate the over
	ov, err := this.Over.Evaluate(item)
	if err != nil {
		return nil, err
	}

	// see if we're dealing with an array
	if ov.Type() == dparval.ARRAY {
		// by accessing the array contents this way
		// we avoid parsing it
		ok := true
		index := 0
		for ok {
			inner, err := ov.Index(index)
			index = index + 1
			if err != nil {
				switch err := err.(type) {
				case *dparval.Undefined:
					ok = false
				default:
					return nil, err
				}
			} else {
				// duplicate the existing context
				innerContext := item.Duplicate()
				// add this object named as the alias
				innerContext.SetPath(this.As, inner)
				if this.Condition != nil {
					// now evaluate the condition in this new context
					innerResult, err := this.Condition.Evaluate(innerContext)
					if err != nil {
						switch err := err.(type) {
						case *dparval.Undefined:
							// this is not true, keep trying
							continue
						default:
							// any other error should be returned to caller
							return nil, err
						}
					}
					innerResultVal := innerResult.Value()
					// check to see if this value is true
					innerBoolResult := ValueInBooleanContext(innerResultVal)
					if innerBoolResult == true {
						// now we have to evaluate the output expression
						outputResult, err := this.Output.Evaluate(innerContext)
						return outputResult, err
					}
				} else {
					// now we have to evaluate the output expression
					outputResult, err := this.Output.Evaluate(innerContext)
					return outputResult, err
				}
			}
		}
		return nil, &dparval.Undefined{}
	}
	return nil, &dparval.Undefined{}
}

func (this *KeyJoin) joinItems(item *dparval.Value, keyItem *dparval.Value) bool {

	if keyItem == nil {
		if this.Type == "LEFT" {
			return this.Base.SendItem(item)
		}
		return true
	}

	newItem := item.Duplicate()
	/* join the item and ship it */
	if this.Projection != nil {
		keyProj, Error := this.Base.Evaluate(this.Projection, keyItem)
		if Error != nil {
			switch err := Error.(type) {
			case *dparval.Undefined:
				return true
			default:
				return this.Base.SendError(query.NewError(err, "Internal error in KeyJoin"))
			}

		}
		newItem.SetPath(this.As, keyProj)
	} else {
		newItem.SetPath(this.As, keyItem)
	}
	this.rowsFetched += 1
	this.Base.SendItem(newItem)
	return true
}

func (this *InterpretedExecutor) processItem(q network.Query, item *dparval.Value) bool {
	projection := item.GetAttachment("projection")
	switch projection := projection.(type) {
	case *dparval.Value:
		result := projection.Value()
		q.Response().SendResult(result)
	}
	return true
}

func (this *Property) Evaluate(item *dparval.Value) (*dparval.Value, error) {
	if item == nil {
		return nil, &dparval.Undefined{this.Path}
	}
	pv, err := item.Path(this.Path)
	if err != nil {
		return nil, err
	}
	return pv, nil
}

func setAggregateValue(item *dparval.Value, key string, val *dparval.Value) {
	aggregates := item.GetAttachment("aggregates")
	aggregatesMap, ok := aggregates.(map[string]interface{})
	if !ok {
		// create a new aggregates map
		aggregatesMap = map[string]interface{}{}
		item.SetAttachment("aggregates", aggregatesMap)
	}
	aggregatesMap[key] = val
}

func (this *FunctionCallSplit) Evaluate(item *dparval.Value) (*dparval.Value, error) {
	// first evaluate the arguments
	av, err := this.Operands[0].Expr.Evaluate(item)

	if err != nil {
		switch err := err.(type) {
		case *dparval.Undefined:
			// undefined returns null
			return dparval.NewValue(nil), nil
		default:
			// any other error return to caller
			return nil, err
		}

		// FIXME warn if arguments were wrong type?
		if av.Type() != dparval.STRING {
			return dparval.NewValue(nil), nil
		}
	}

	var sep *dparval.Value = nil
	if len(this.Operands) > 1 {
		sep, err = this.Operands[1].Expr.Evaluate(item)
		if err != nil {
			switch err := err.(type) {
			case *dparval.Undefined:
				// undefined returns null
				return dparval.NewValue(nil), nil
			default:
				// any other error return to caller
				return nil, err
			}
		}

		// FIXME warn if arguments were wrong type?
		if sep.Type() != dparval.STRING {
			return dparval.NewValue(nil), nil
		}
	}

	var sa []string
	if sep != nil {
		sa = strings.Split(av.Value().(string),
			sep.Value().(string))
	} else {
		sa = strings.Fields(av.Value().(string))
	}

	rv := make([]interface{}, len(sa))
	for i, s := range sa {
		rv[i] = s
	}

	return dparval.NewValue(rv), nil
}

func (this *EliminateDuplicates) processItem(item *dparval.Value) bool {
	projection, _ := item.GetAttachment("projection").(*dparval.Value)
	if projection != nil {
		hashval := fmt.Sprintf("%s", projection.Value())
		found := this.uniqueBuffer[hashval]
		if found == nil {
			this.uniqueBuffer[hashval] = item
		}
	}
	return true
}

func (this *FunctionCallNowMillis) Evaluate(item *dparval.Value) (*dparval.Value, error) {
	// first retrieve the query object
	q := item.GetAttachment("query")

	query, ok := q.(network.Query)
	if !ok {
		return dparval.NewValue(nil), nil
	}

	return dparval.NewValue(timeToMillis(query.StartTime())), nil
}

func eliminateNonObject(val *dparval.Value, err error) (*dparval.Value, error) {
	val, err = eliminateNullMissing(val, err)
	if err != nil {
		return nil, err
	}
	if val != nil {
		if val.Type() == dparval.OBJECT {
			return val, nil
		}
	}
	return nil, nil
}

func eliminateNonNumber(val *dparval.Value, err error) (*dparval.Value, error) {
	val, err = eliminateNullMissing(val, err)
	if err != nil {
		return nil, err
	}
	if val != nil {
		if val.Type() == dparval.NUMBER {
			return val, nil
		}
	}
	return nil, nil
}

// lookup the current value of the aggregate stored
// at the specified key
func aggregateValue(item *dparval.Value, key string) (*dparval.Value, error) {
	aggregates := item.GetAttachment("aggregates")
	aggregatesMap, ok := aggregates.(map[string]interface{})
	if ok {
		value := aggregatesMap[key]
		valuedparval, ok := value.(*dparval.Value)
		if ok {
			return valuedparval, nil
		}
	}
	return nil, fmt.Errorf("Unable to find aggregate %s", key)
}

func eliminateNullMissing(val *dparval.Value, err error) (*dparval.Value, error) {
	if err != nil {
		switch err := err.(type) {
		case *dparval.Undefined:
			// missing elimination
			return nil, nil
		default:
			return nil, err
		}
	}
	if val.Type() == dparval.NULL {
		// null elimination
		return nil, nil
	}
	return val, nil
}

func (this *Project) processItem(item *dparval.Value) bool {
	resultMap := map[string]interface{}{}
	for _, resultItem := range this.Result {

		val, err := this.Base.projectedValueOfResultExpression(item, resultItem)
		if err != nil {
			switch err := err.(type) {
			case *dparval.Undefined:
				// undefined contributes nothing to the result map
				continue
			default:
				return this.Base.SendError(query.NewError(err, "unexpected error projecting expression"))
			}
		}

		if resultItem.Star {
			if val != nil {
				valval := val.Value()
				switch valval := valval.(type) {
				case map[string]interface{}:
					// then if the result was an object
					// add its contents ot the result map
					for k, v := range valval {
						resultMap[k] = v
					}
				}
			}
		} else {
			resultMap[resultItem.As] = val
		}
	}

	if !this.projectEmpty && len(resultMap) == 0 {
		return true
	}

	// build a Value from the projection
	projection := dparval.NewValue(resultMap)

	// store the projection as an attachment on the main item
	item.SetAttachment("projection", projection)

	clog.To(DEBUG_PROJECT_CHANNEL, "projecting: %v", item)

	// write to the output
	return this.Base.SendItem(item)
}

func (this *Order) processItem(item *dparval.Value) bool {
	if this.explicitAliases != nil {
		projection := item.GetAttachment("projection")
		projectionValue, ok := projection.(*dparval.Value)
		if ok {
			for _, explicitAlias := range this.explicitAliases {
				// put the explicit alias values from the projection into the item
				aliasedProjectedValue, err := projectionValue.Path(explicitAlias)
				if err == nil {
					item.SetPath(explicitAlias, aliasedProjectedValue)
				}
			}
		}
	}
	this.buffer = append(this.buffer, item)
	return true
}

func (this *KeyNest) flushBatch(baseItem *dparval.Value, ids []string) bool {

	bulkResponse, err := this.bucket.BulkFetch(ids)
	if err != nil {
		return this.Base.SendError(query.NewError(err, "error getting bulk response"))
	}

	// now we need to emit the bulk fetched items in the correct order (from the id list)

	for _, v := range ids {
		item, ok := bulkResponse[v]

		if ok {
			if this.Projection != nil {
				projectedVal, err := this.Base.projectedValueOfResultExpression(item, ast.NewResultExpression(this.Projection))
				if err != nil {
					switch err := err.(type) {
					case *dparval.Undefined:
						// undefined contributes nothing to the result map
						continue
					default:
						return this.Base.SendError(query.NewError(err, "unexpected error projecting fetch expression"))
					}
				} else {
					this.Right = append(this.Right, projectedVal)
				}
			} else {
				this.Right = append(this.Right, item)
			}
			this.rowsFetched += 1
		}
	}
	if len(this.Right) > 0 {
		baseItem.SetPath(this.As, this.Right)
	}
	// if the lenght of the array is 0 and the type of join is not LEFT
	// then we return an empty result for this evaluation
	if len(this.Right) == 0 && this.Type != "LEFT" {
		return true
	}

	this.Base.SendItem(baseItem)

	return true
}

func (this *Unnest) processItem(item *dparval.Value) bool {

	val, err := this.Base.Evaluate(this.Over, item)
	if err != nil {
		switch err := err.(type) {
		case *dparval.Undefined:
			if val == nil && this.Type == "LEFT" {
				return this.Base.SendItem(item)
			}
			return true
		default:
			return this.Base.SendError(query.NewError(err, "Internal Error"))
		}
	}

	if val.Type() == dparval.ARRAY {
		ok := true
		index := 0
		for ok {
			inner, err := val.Index(index)
			index = index + 1
			if err != nil {
				switch err := err.(type) {
				case *dparval.Undefined:
					ok = false
				default:
					this.Base.SendError(query.NewError(err, "Internal Error"))
					return false
				}
			} else {
				newItem := item.Duplicate()
				newItem.SetPath(this.As, inner)
				this.Base.SendItem(newItem)
			}
		}
	} else if this.Type == "LEFT" {
		// send back the item since this is a left join
		this.Base.SendItem(item)
	}

	return true
}

func (this *KeyJoin) flushBatch(baseItem *dparval.Value, ids []string) bool {

	bulkResponse, err := this.bucket.BulkFetch(ids)
	if err != nil {
		return this.Base.SendError(query.NewError(err, "error getting bulk response"))
	}

	// now we need to emit the bulk fetched items in the correct order (from the id list)

	for _, v := range ids {
		item, ok := bulkResponse[v]
		newItem := baseItem.Duplicate()
		if item == nil {
			if this.Type == "LEFT" {
				this.Base.SendItem(newItem)
			}
			continue
		}
		if ok {
			if this.Projection != nil {
				projectedVal, err := this.Base.projectedValueOfResultExpression(item, ast.NewResultExpression(this.Projection))
				if err != nil {
					switch err := err.(type) {
					case *dparval.Undefined:
						// undefined contributes nothing to the result map
						continue
					default:
						return this.Base.SendError(query.NewError(err, "unexpected error projecting fetch expression"))
					}
				} else {
					newItem.SetPath(this.As, projectedVal)
				}
			} else {
				newItem.SetPath(this.As, item)
			}
			this.Base.SendItem(newItem)
			this.rowsFetched += 1
		}
	}

	return true
}

func (this *CaseOperator) Evaluate(item *dparval.Value) (*dparval.Value, error) {

	var switchItem *dparval.Value

	if this.Switch != nil {
		var err error
		switchItem, err = this.Switch.Evaluate(item)
		if err != nil {
			return nil, err
		}
	}

	// walk through the WhenThens in order
	for _, WhenThen := range this.WhenThens {
		// evalute the when
		whenVal, err := WhenThen.When.Evaluate(item)
		if err != nil {
			switch err := err.(type) {
			case *dparval.Undefined:
				// MISSING is not true, keep looking
				continue
			default:
				// any other error should be returned to caller
				return nil, err
			}
		}

		var match bool
		whenValVal := whenVal.Value()

		if switchItem != nil {
			if switchItem.Value() == whenValVal {
				match = true
			}
		} else {

			match = ValueInBooleanContext(whenValVal).(bool)
		}

		if match == true {
			// evalauate the then
			thenVal, err := WhenThen.Then.Evaluate(item)
			if err != nil {
				return nil, err
			}
			return thenVal, nil
		}
	}

	// if we got this point, none of the WhenThen pairs were satisfied
	// if there was an else, evaluate it and return
	if this.Else != nil {
		elseVal, err := this.Else.Evaluate(item)
		if err != nil {
			return nil, err
		}
		return elseVal, nil
	}
	// if there was no else, return null
	return dparval.NewValue(nil), nil
}

func (this *KeyNest) processItem(item *dparval.Value) bool {

	if item == nil {
		return true
	}

	this.Right = make([]interface{}, 0)
	newItem := item.Duplicate()

	val, err := this.Base.Evaluate(this.Keys.Expr, item)
	if err != nil {
		switch err := err.(type) {
		case *dparval.Undefined:
			if this.Type == "LEFT" {
				this.Base.SendItem(newItem)
				return true
			}
			return true
		default:
			return this.Base.SendError(query.NewError(err, "Internal error in KeyNest"))
		}
	}

	if val.Type() == dparval.STRING {
		if this.Keys.Type == "KEYS" {
			this.Base.SendError(query.NewError(fmt.Errorf("KEYS expression should evaluate to an array"), ""))
			return false
		}
		if val.Value() == nil {
			if this.Type == "LEFT" {
				return this.Base.SendItem(item)
			}
			return true
		}

		fetch_id := val.Value().(string)
		keyitem, err := this.bucket.Fetch(fetch_id)
		if err != nil {
			this.Base.SendError(query.NewError(err, "Unable to fetch key"))
			return false
		}

		if keyitem != nil {
			if this.Projection != nil {
				keyproj, Error := this.Base.Evaluate(this.Projection, keyitem)
				if Error != nil {
					switch err := Error.(type) {
					case *dparval.Undefined:
						return true
					default:
						return this.Base.SendError(query.NewError(err, "Internal error in KeyNest"))
					}

				}
				this.Right = append(this.Right, keyproj)
			} else {
				this.Right = append(this.Right, keyitem)
			}

			newItem.SetPath(this.As, this.Right)
			this.Base.SendItem(newItem)
		} else if this.Type == "LEFT" {

			newItem.SetPath(this.As, this.Right)
			this.Base.SendItem(newItem)
		}

		this.rowsFetched += 1

	} else if val.Type() == dparval.ARRAY {
		ok := true
		index := 0

		ids := make([]string, 0, FETCH_BATCH_SIZE)

		if this.Keys.Type == "KEY" {
			this.Base.SendError(query.NewError(fmt.Errorf("KEY used with an array argument"), ""))
			return false
		}

		for ok {
			id, err := val.Index(index)
			index = index + 1

			array_len := len(val.Value().([]interface{}))
			if array_len == 0 {
				if this.Type == "LEFT" {
					item.SetPath(this.As, this.Right)
					this.Base.SendItem(item)
				}
				return true
			}

			if err != nil {
				if len(ids) != 0 {
					return this.flushBatch(item, ids)
				}
				return true
			}

//.........这里部分代码省略.........

func (this *BaseOperator) Evaluate(e ast.Expression, item *dparval.Value) (*dparval.Value, error) {
	// first ensure the query is avaliable
	item.SetAttachment("query", this.query)
	return e.Evaluate(item)
}