text: Title: [BUG] NameError raised when comparing string inside agg function.
  Body: <!--
  Thank you for your contribution!
  
  Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
  -->
  
  **Describe the bug**
  
  NameError raised when comparing string inside agg function.
  **To Reproduce**
  To help us reproducing this bug, please provide information below:
  1. Your Python version
  2. The version of Mars you use
  3. Versions of crucial packages, such as numpy, scipy and pandas
  4. Full stack of the error.
  5. Minimized code to reproduce the error.
  
  ```
  In [3]: import mars.dataframe as md
  
  In [4]: df = md.DataFrame({'a': ['1', '2', '3'], 'b': ['a1', 'a2', 'a1']})
  
  In [5]: df.groupby('a', as_index=False)['b'].agg(lambda x: (x == "a1").sum()).ex
     ...: ecute()
    0%|                                                   | 0/100 [00:00<?, ?it/s]Failed to run subtask aRWmQqEgc5OymCkUZLYRcTb9 on band numa-0
  Traceback (most recent call last):
    File "/Users/xuyeqin/Workspace/mars/mars/services/scheduling/worker/execution.py", line 315, in internal_run_subtask
      subtask_info.result = await self._retry_run_subtask(
    File "/Users/xuyeqin/Workspace/mars/mars/services/scheduling/worker/execution.py", line 404, in _retry_run_subtask
      return await _retry_run(subtask, subtask_info, _run_subtask_once)
    File "/Users/xuyeqin/Workspace/mars/mars/services/scheduling/worker/execution.py", line 91, in _retry_run
      raise ex
    File "/Users/xuyeqin/Workspace/mars/mars/services/scheduling/worker/execution.py", line 66, in _retry_run
      return await target_async_func(*args)
    File "/Users/xuyeqin/Workspace/mars/mars/services/scheduling/worker/execution.py", line 357, in _run_subtask_once
      return await asyncio.shield(aiotask)
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/api.py", line 66, in run_subtask_in_slot
      return await ref.run_subtask.options(profiling_context=profiling_context).send(
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/context.py", line 186, in send
      return self._process_result_message(result)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/context.py", line 70, in _process_result_message
      raise message.error.with_traceback(message.traceback)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/pool.py", line 520, in send
      result = await self._run_coro(message.message_id, coro)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/pool.py", line 319, in _run_coro
      return await coro
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/api.py", line 115, in __on_receive__
      return await super().__on_receive__(message)
    File "mars/oscar/core.pyx", line 373, in __on_receive__
      raise ex
    File "mars/oscar/core.pyx", line 367, in mars.oscar.core._BaseActor.__on_receive__
      return await self._handle_actor_result(result)
    File "mars/oscar/core.pyx", line 252, in _handle_actor_result
      task_result = await coros[0]
    File "mars/oscar/core.pyx", line 295, in _run_actor_async_generator
      with debug_async_timeout('actor_lock_timeout',
    File "mars/oscar/core.pyx", line 297, in mars.oscar.core._BaseActor._run_actor_async_generator
      async with self._lock:
    File "mars/oscar/core.pyx", line 301, in mars.oscar.core._BaseActor._run_actor_async_generator
      res = await gen.athrow(*res)
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/worker/runner.py", line 118, in run_subtask
      result = yield self._running_processor.run(subtask)
    File "mars/oscar/core.pyx", line 306, in mars.oscar.core._BaseActor._run_actor_async_generator
      res = await self._handle_actor_result(res)
    File "mars/oscar/core.pyx", line 226, in _handle_actor_result
      result = await result
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/context.py", line 186, in send
      return self._process_result_message(result)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/context.py", line 70, in _process_result_message
      raise message.error.with_traceback(message.traceback)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/pool.py", line 520, in send
      result = await self._run_coro(message.message_id, coro)
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/backends/pool.py", line 319, in _run_coro
      return await coro
    File "/Users/xuyeqin/Workspace/mars/mars/oscar/api.py", line 115, in __on_receive__
      return await super().__on_receive__(message)
    File "mars/oscar/core.pyx", line 373, in __on_receive__
      raise ex
    File "mars/oscar/core.pyx", line 367, in mars.oscar.core._BaseActor.__on_receive__
      return await self._handle_actor_result(result)
    File "mars/oscar/core.pyx", line 252, in _handle_actor_result
      task_result = await coros[0]
    File "mars/oscar/core.pyx", line 295, in _run_actor_async_generator
      with debug_async_timeout('actor_lock_timeout',
    File "mars/oscar/core.pyx", line 297, in mars.oscar.core._BaseActor._run_actor_async_generator
      async with self._lock:
    File "mars/oscar/core.pyx", line 301, in mars.oscar.core._BaseActor._run_actor_async_generator
      res = await gen.athrow(*res)
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/worker/processor.py", line 596, in run
      result = yield self._running_aio_task
    File "mars/oscar/core.pyx", line 306, in mars.oscar.core._BaseActor._run_actor_async_generator
      res = await self._handle_actor_result(res)
    File "mars/oscar/core.pyx", line 226, in _handle_actor_result
      result = await result
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/worker/processor.py", line 457, in run
      await self._execute_graph(chunk_graph)
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/worker/processor.py", line 209, in _execute_graph
      await to_wait
    File "/Users/xuyeqin/Workspace/mars/mars/lib/aio/_threads.py", line 36, in to_thread
      return await loop.run_in_executor(None, func_call)
    File "/Users/xuyeqin/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/thread.py", line 57, in run
      result = self.fn(*self.args, **self.kwargs)
    File "/Users/xuyeqin/Workspace/mars/mars/core/mode.py", line 77, in _inner
      return func(*args, **kwargs)
    File "/Users/xuyeqin/Workspace/mars/mars/services/subtask/worker/processor.py", line 177, in _execute_operand
      return execute(ctx, op)
    File "/Users/xuyeqin/Workspace/mars/mars/core/operand/core.py", line 487, in execute
      result = executor(results, op)
    File "/Users/xuyeqin/Workspace/mars/mars/core/custom_log.py", line 94, in wrap
      return func(cls, ctx, op)
    File "/Users/xuyeqin/Workspace/mars/mars/utils.py", line 1128, in wrapped
      result = func(cls, ctx, op)
    File "/Users/xuyeqin/Workspace/mars/mars/dataframe/groupby/aggregation.py", line 1050, in execute
      cls._execute_map(ctx, op)
    File "/Users/xuyeqin/Workspace/mars/mars/dataframe/groupby/aggregation.py", line 854, in _execute_map
      pre_df = func(pre_df, gpu=op.is_gpu())
    File "<string>", line 3, in expr_function
  NameError: name 'a1' is not defined
    0%|                                                   | 0/100 [00:00<?, ?it/s]
  ---------------------------------------------------------------------------
  NameError                                 Traceback (most recent call last)
  <ipython-input-5-3a632fccd78f> in <module>
  ----> 1 df.groupby('a', as_index=False)['b'].agg(lambda x: (x == "a1").sum()).execute()
  
  ~/Workspace/mars/mars/core/entity/tileables.py in execute(self, session, **kw)
      460 
      461     def execute(self, session=None, **kw):
  --> 462         result = self.data.execute(session=session, **kw)
      463         if isinstance(result, TILEABLE_TYPE):
      464             return self
  
  ~/Workspace/mars/mars/core/entity/executable.py in execute(self, session, **kw)
       96 
       97         session = _get_session(self, session)
  ---> 98         return execute(self, session=session, **kw)
       99 
      100     def _check_session(self, session: SessionType, action: str):
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in execute(tileable, session, wait, new_session_kwargs, show_progress, progress_update_interval, *tileables, **kwargs)
     1771         session = get_default_or_create(**(new_session_kwargs or dict()))
     1772     session = _ensure_sync(session)
  -> 1773     return session.execute(
     1774         tileable,
     1775         *tileables,
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in execute(self, tileable, show_progress, *tileables, **kwargs)
     1571         fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
     1572         try:
  -> 1573             execution_info: ExecutionInfo = fut.result(
     1574                 timeout=self._isolated_session.timeout
     1575             )
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/_base.py in result(self, timeout)
      437                 raise CancelledError()
      438             elif self._state == FINISHED:
  --> 439                 return self.__get_result()
      440             else:
      441                 raise TimeoutError()
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/_base.py in __get_result(self)
      386     def __get_result(self):
      387         if self._exception:
  --> 388             raise self._exception
      389         else:
      390             return self._result
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in _execute(session, wait, show_progress, progress_update_interval, cancelled, *tileables, **kwargs)
     1722                 while not cancelled.is_set():
     1723                     try:
  -> 1724                         await asyncio.wait_for(
     1725                             asyncio.shield(execution_info), progress_update_interval
     1726                         )
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/asyncio/tasks.py in wait_for(fut, timeout, loop)
      481 
      482         if fut.done():
  --> 483             return fut.result()
      484         else:
      485             fut.remove_done_callback(cb)
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/asyncio/tasks.py in _wrap_awaitable(awaitable)
      682     that will later be wrapped in a Task by ensure_future().
      683     """
  --> 684     return (yield from awaitable.__await__())
      685 
      686 _wrap_awaitable._is_coroutine = _is_coroutine
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in wait()
      100 
      101             async def wait():
  --> 102                 return await self._aio_task
      103 
      104             self._future_local.future = fut = asyncio.run_coroutine_threadsafe(
  
  ~/Workspace/mars/mars/deploy/oscar/session.py in _run_in_background(self, tileables, task_id, progress, profiling)
      901                     )
      902                 if task_result.error:
  --> 903                     raise task_result.error.with_traceback(task_result.traceback)
      904             if cancelled:
      905                 return
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in internal_run_subtask(self, subtask, band_name)
      313 
      314             batch_quota_req = {(subtask.session_id, subtask.subtask_id): calc_size}
  --> 315             subtask_info.result = await self._retry_run_subtask(
      316                 subtask, band_name, subtask_api, batch_quota_req
      317             )
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run_subtask(self, subtask, band_name, subtask_api, batch_quota_req)
      402         #  any exceptions occurred.
      403         if subtask.retryable:
  --> 404             return await _retry_run(subtask, subtask_info, _run_subtask_once)
      405         else:
      406             return await _run_subtask_once()
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run(subtask, subtask_info, target_async_func, *args)
       89                     ex,
       90                 )
  ---> 91             raise ex
       92 
       93 
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _retry_run(subtask, subtask_info, target_async_func, *args)
       64     while True:
       65         try:
  ---> 66             return await target_async_func(*args)
       67         except (OSError, MarsError) as ex:
       68             if subtask_info.num_retries < subtask_info.max_retries:
  
  ~/Workspace/mars/mars/services/scheduling/worker/execution.py in _run_subtask_once()
      355                     subtask_api.run_subtask_in_slot(band_name, slot_id, subtask)
      356                 )
  --> 357                 return await asyncio.shield(aiotask)
      358             except asyncio.CancelledError as ex:
      359                 # make sure allocated slots are traced
  
  ~/Workspace/mars/mars/services/subtask/api.py in run_subtask_in_slot(self, band_name, slot_id, subtask)
       64             else None
       65         )
  ---> 66         return await ref.run_subtask.options(profiling_context=profiling_context).send(
       67             subtask
       68         )
  
  ~/Workspace/mars/mars/oscar/backends/context.py in send(self, actor_ref, message, wait_response, profiling_context)
      184             if wait_response:
      185                 result = await self._wait(future, actor_ref.address, message)
  --> 186                 return self._process_result_message(result)
      187             else:
      188                 return future
  
  ~/Workspace/mars/mars/oscar/backends/context.py in _process_result_message(message)
       68             return message.result
       69         else:
  ---> 70             raise message.error.with_traceback(message.traceback)
       71 
       72     async def _wait(self, future: asyncio.Future, address: str, message: _MessageBase):
  
  ~/Workspace/mars/mars/oscar/backends/pool.py in send()
      518                 raise ActorNotExist(f"Actor {actor_id} does not exist")
      519             coro = self._actors[actor_id].__on_receive__(message.content)
  --> 520             result = await self._run_coro(message.message_id, coro)
      521             processor.result = ResultMessage(
      522                 message.message_id,
  
  ~/Workspace/mars/mars/oscar/backends/pool.py in _run_coro()
      317         self._process_messages[message_id] = asyncio.tasks.current_task()
      318         try:
  --> 319             return await coro
      320         finally:
      321             self._process_messages.pop(message_id, None)
  
  ~/Workspace/mars/mars/oscar/api.py in __on_receive__()
      113             Message shall be (method_name,) + args + (kwargs,)
      114         """
  --> 115         return await super().__on_receive__(message)
      116 
      117 
  
  ~/Workspace/mars/mars/oscar/core.pyx in __on_receive__()
      371                 debug_logger.exception('Got unhandled error when handling message %r'
      372                                        'in actor %s at %s', message, self.uid, self.address)
  --> 373             raise ex
      374 
      375 
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor.__on_receive__()
      365                 raise ValueError(f'call_method {call_method} not valid')
      366 
  --> 367             return await self._handle_actor_result(result)
      368         except Exception as ex:
      369             if _log_unhandled_errors:
  
  ~/Workspace/mars/mars/oscar/core.pyx in _handle_actor_result()
      250             # asyncio.wait as it introduces much overhead
      251             if len(coros) == 1:
  --> 252                 task_result = await coros[0]
      253                 if extract_tuple:
      254                     result = task_result
  
  ~/Workspace/mars/mars/oscar/core.pyx in _run_actor_async_generator()
      293             res = None
      294             while True:
  --> 295                 with debug_async_timeout('actor_lock_timeout',
      296                                          'async_generator %r hold lock timeout', gen):
      297                     async with self._lock:
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      295                 with debug_async_timeout('actor_lock_timeout',
      296                                          'async_generator %r hold lock timeout', gen):
  --> 297                     async with self._lock:
      298                         if not is_exception:
      299                             res = await gen.asend(res)
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      299                             res = await gen.asend(res)
      300                         else:
  --> 301                             res = await gen.athrow(*res)
      302                 try:
      303                     if _log_cycle_send:
  
  ~/Workspace/mars/mars/services/subtask/worker/runner.py in run_subtask()
      116         self._running_processor = self._last_processor = processor
      117         try:
  --> 118             result = yield self._running_processor.run(subtask)
      119         finally:
      120             self._running_processor = None
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      304                         message_trace = pop_message_trace()
      305 
  --> 306                     res = await self._handle_actor_result(res)
      307                     is_exception = False
      308                 except:
  
  ~/Workspace/mars/mars/oscar/core.pyx in _handle_actor_result()
      224 
      225         if inspect.isawaitable(result):
  --> 226             result = await result
      227         elif is_async_generator(result):
      228             result = (result,)
  
  ~/Workspace/mars/mars/oscar/backends/context.py in send()
      184             if wait_response:
      185                 result = await self._wait(future, actor_ref.address, message)
  --> 186                 return self._process_result_message(result)
      187             else:
      188                 return future
  
  ~/Workspace/mars/mars/oscar/backends/context.py in _process_result_message()
       68             return message.result
       69         else:
  ---> 70             raise message.error.with_traceback(message.traceback)
       71 
       72     async def _wait(self, future: asyncio.Future, address: str, message: _MessageBase):
  
  ~/Workspace/mars/mars/oscar/backends/pool.py in send()
      518                 raise ActorNotExist(f"Actor {actor_id} does not exist")
      519             coro = self._actors[actor_id].__on_receive__(message.content)
  --> 520             result = await self._run_coro(message.message_id, coro)
      521             processor.result = ResultMessage(
      522                 message.message_id,
  
  ~/Workspace/mars/mars/oscar/backends/pool.py in _run_coro()
      317         self._process_messages[message_id] = asyncio.tasks.current_task()
      318         try:
  --> 319             return await coro
      320         finally:
      321             self._process_messages.pop(message_id, None)
  
  ~/Workspace/mars/mars/oscar/api.py in __on_receive__()
      113             Message shall be (method_name,) + args + (kwargs,)
      114         """
  --> 115         return await super().__on_receive__(message)
      116 
      117 
  
  ~/Workspace/mars/mars/oscar/core.pyx in __on_receive__()
      371                 debug_logger.exception('Got unhandled error when handling message %r'
      372                                        'in actor %s at %s', message, self.uid, self.address)
  --> 373             raise ex
      374 
      375 
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor.__on_receive__()
      365                 raise ValueError(f'call_method {call_method} not valid')
      366 
  --> 367             return await self._handle_actor_result(result)
      368         except Exception as ex:
      369             if _log_unhandled_errors:
  
  ~/Workspace/mars/mars/oscar/core.pyx in _handle_actor_result()
      250             # asyncio.wait as it introduces much overhead
      251             if len(coros) == 1:
  --> 252                 task_result = await coros[0]
      253                 if extract_tuple:
      254                     result = task_result
  
  ~/Workspace/mars/mars/oscar/core.pyx in _run_actor_async_generator()
      293             res = None
      294             while True:
  --> 295                 with debug_async_timeout('actor_lock_timeout',
      296                                          'async_generator %r hold lock timeout', gen):
      297                     async with self._lock:
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      295                 with debug_async_timeout('actor_lock_timeout',
      296                                          'async_generator %r hold lock timeout', gen):
  --> 297                     async with self._lock:
      298                         if not is_exception:
      299                             res = await gen.asend(res)
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      299                             res = await gen.asend(res)
      300                         else:
  --> 301                             res = await gen.athrow(*res)
      302                 try:
      303                     if _log_cycle_send:
  
  ~/Workspace/mars/mars/services/subtask/worker/processor.py in run()
      594         self._running_aio_task = asyncio.create_task(processor.run())
      595         try:
  --> 596             result = yield self._running_aio_task
      597             raise mo.Return(result)
      598         finally:
  
  ~/Workspace/mars/mars/oscar/core.pyx in mars.oscar.core._BaseActor._run_actor_async_generator()
      304                         message_trace = pop_message_trace()
      305 
  --> 306                     res = await self._handle_actor_result(res)
      307                     is_exception = False
      308                 except:
  
  ~/Workspace/mars/mars/oscar/core.pyx in _handle_actor_result()
      224 
      225         if inspect.isawaitable(result):
  --> 226             result = await result
      227         elif is_async_generator(result):
      228             result = (result,)
  
  ~/Workspace/mars/mars/services/subtask/worker/processor.py in run()
      455             try:
      456                 # execute chunk graph
  --> 457                 await self._execute_graph(chunk_graph)
      458             finally:
      459                 # unpin inputs data
  
  ~/Workspace/mars/mars/services/subtask/worker/processor.py in _execute_graph()
      207 
      208                 try:
  --> 209                     await to_wait
      210                     logger.debug(
      211                         "Finish executing operand: %s," "chunk: %s, subtask id: %s",
  
  ~/Workspace/mars/mars/lib/aio/_threads.py in to_thread()
       34     ctx = contextvars.copy_context()
       35     func_call = functools.partial(ctx.run, func, *args, **kwargs)
  ---> 36     return await loop.run_in_executor(None, func_call)
  
  ~/miniconda3/envs/mars3.8/lib/python3.8/concurrent/futures/thread.py in run()
       55 
       56         try:
  ---> 57             result = self.fn(*self.args, **self.kwargs)
       58         except BaseException as exc:
       59             self.future.set_exception(exc)
  
  ~/Workspace/mars/mars/core/mode.py in _inner()
       75             def _inner(*args, **kwargs):
       76                 with enter_mode(**mode_name_to_value):
  ---> 77                     return func(*args, **kwargs)
       78 
       79         else:
  
  ~/Workspace/mars/mars/services/subtask/worker/processor.py in _execute_operand()
      175         self, ctx: Dict[str, Any], op: OperandType
      176     ):  # noqa: R0201  # pylint: disable=no-self-use
  --> 177         return execute(ctx, op)
      178 
      179     async def _execute_graph(self, chunk_graph: ChunkGraph):
  
  ~/Workspace/mars/mars/core/operand/core.py in execute()
      485             # The `UFuncTypeError` was introduced by numpy#12593 since v1.17.0.
      486             try:
  --> 487                 result = executor(results, op)
      488                 succeeded = True
      489                 return result
  
  ~/Workspace/mars/mars/core/custom_log.py in wrap()
       92 
       93         if custom_log_dir is None:
  ---> 94             return func(cls, ctx, op)
       95 
       96         log_path = os.path.join(custom_log_dir, op.key)
  
  ~/Workspace/mars/mars/utils.py in wrapped()
     1126 
     1127         try:
  -> 1128             result = func(cls, ctx, op)
     1129         finally:
     1130             with AbstractSession._lock:
  
  ~/Workspace/mars/mars/dataframe/groupby/aggregation.py in execute()
     1048             pd.set_option("mode.use_inf_as_na", op.use_inf_as_na)
     1049             if op.stage == OperandStage.map:
  -> 1050                 cls._execute_map(ctx, op)
     1051             elif op.stage == OperandStage.combine:
     1052                 cls._execute_combine(ctx, op)
  
  ~/Workspace/mars/mars/dataframe/groupby/aggregation.py in _execute_map()
      852                 pre_df = in_data if cols is None else in_data[cols]
      853                 try:
  --> 854                     pre_df = func(pre_df, gpu=op.is_gpu())
      855                 except TypeError:
      856                     pre_df = pre_df.transform(_wrapped_func)
  
  ~/Workspace/mars/mars/dataframe/reduction/core.py in expr_function()
  
  NameError: name 'a1' is not defined
  ```
  