code: func (cf *countFunction) StreamUpdate(row []types.Datum, ectx context.Context) error {
  	ctx := cf.getStreamedContext()
  	var vals []interface{}
  	if cf.Distinct {
  		vals = make([]interface{}, 0, len(cf.Args))
  	}
  	for _, a := range cf.Args {
  		value, err := a.Eval(row, ectx)
  		if err != nil {
  			return errors.Trace(err)
  		}
  		if value.GetValue() == nil {
  			return nil
  		}
  		if cf.Distinct {
  			vals = append(vals, value.GetValue())
  		}
  	}
  	if cf.Distinct {
  		d, err := ctx.DistinctChecker.Check(vals)
  		if err != nil {
  			return errors.Trace(err)
  		}
  		if !d {
  			return nil
  		}
  	}
  	ctx.Count++
  	return nil
  }
docs: // StreamUpdate implements AggregationFunction interface.