[BUG] `df.apply` failed when unknown chunks exist on the apply axis.

Nobody has claimed this yet.

**Describe the bug**

`df.apply` failed when unknown chunks exist on the apply axis.

**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 [1]: import mars.tensor as mt

In [2]: import mars.dataframe as md

In [3]: df = md.DataFrame(mt.random.rand(10, 3), chunk_size=5)

In [4]: df = df[df[0] < 0.9]

In [5]: df.apply(lambda x: x).execute()
0%| | 0/100 [00:00 in
----> 1 df.apply(lambda x: x).execute()

~/Workspace/mars/mars/core/entity/tileables.py in execute(self, session, **kw)
442
443 def execute(self, session=None, **kw):
--> 444 result = self.data.execute(session=session, **kw)
445 if isinstance(result, TILEABLE_TYPE):
446 return self

~/Workspace/mars/mars/core/entity/executable.py in execute(self, session, **kw)
102
103 session = _get_session(self, session)
--> 104 return execute(self, session=session, **kw)
105
106 def _check_session(self,

~/Workspace/mars/mars/deploy/oscar/session.py in execute(tileable, session, wait, new_session_kwargs, show_progress, progress_update_interval, *tileables, **kwargs)
1365 **(new_session_kwargs or dict()))
1366 session = _ensure_sync(session)
-> 1367 return session.execute(tileable, *tileables, wait=wait,
1368 show_progress=show_progress,
1369 progress_update_interval=progress_update_interval,

~/Workspace/mars/mars/deploy/oscar/session.py in execute(self, tileable, show_progress, *tileables, **kwargs)
1209 fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
1210 try:
-> 1211 execution_info: ExecutionInfo = fut.result(
1212 timeout=self._isolated_session.timeout)
1213 except KeyboardInterrupt: # pragma: no cover

~/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)
1323 while not cancelled.is_set():
1324 try:
-> 1325 await asyncio.wait_for(asyncio.shield(execution_info),
1326 progress_update_interval)
1327 # done

~/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()
75 except AttributeError:
76 async def wait():
---> 77 return await self._aio_task
78
79 self._future_local.future = fut = \

~/Workspace/mars/mars/deploy/oscar/session.py in _run_in_background(self, tileables, task_id, progress)
689 raise TimeoutError(f'Task({task_id}) running time > {self.timeout}')
690 if task_result.error:
--> 691 raise task_result.error.with_traceback(task_result.traceback)
692 if cancelled:
693 return

~/Workspace/mars/mars/services/task/supervisor/processor.py in inner(processor, *args, **kwargs)
44 async def inner(processor: "TaskProcessor", *args, **kwargs):
45 try:
---> 46 return await func(processor, *args, **kwargs)
47 except: # noqa: E722 # nosec # pylint: disable=bare-except # pragma: no cover
48 processor._err_infos.append(sys.exc_info())

~/Workspace/mars/mars/services/task/supervisor/processor.py in get_next_stage_processor(self)
247 self._init_chunk_graph_iter(tileable_graph)
248
--> 249 chunk_graph = await self._get_next_chunk_graph(self._chunk_graph_iter)
250 if chunk_graph is None:
251 # tile finished

~/Workspace/mars/mars/services/task/supervisor/processor.py in _get_next_chunk_graph(self, chunk_graph_iter)
225
226 fut = asyncio.to_thread(next_chunk_graph)
--> 227 chunk_graph = await fut
228 return chunk_graph
229

~/Workspace/mars/mars/lib/aio/_threads.py in to_thread(func, *args, **kwargs)
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(self)
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/services/task/supervisor/processor.py in next_chunk_graph()
220 def next_chunk_graph():
221 try:
--> 222 return next(chunk_graph_iter)
223 except StopIteration:
224 return

~/Workspace/mars/mars/services/task/supervisor/preprocessor.py in tile(self, tileable_graph)
136 optimize = self._config.optimize_chunk_graph
137 meta_updated = set()
--> 138 for chunk_graph in chunk_graph_builder.build():
139 # optimize chunk graph
140 if optimize:

~/Workspace/mars/mars/core/graph/builder/chunk.py in build(self)
237 def build(self) -> Generator[Union[TileableGraph, ChunkGraph], None, None]:
238 with enter_mode(build=True, kernel=True):
--> 239 yield from self._build()

~/Workspace/mars/mars/core/graph/builder/chunk.py in _build(self)
233
234 def _build(self) -> Iterable[Union[TileableGraph, ChunkGraph]]:
--> 235 yield from self.tiler
236
237 def build(self) -> Generator[Union[TileableGraph, ChunkGraph], None, None]:

~/Workspace/mars/mars/services/task/supervisor/preprocessor.py in __iter__(self)
64 def __iter__(self):
65 while self._tileable_handlers:
---> 66 to_update_tileables = self._iter()
67 if not self.cancelled:
68 yield self._cur_chunk_graph

~/Workspace/mars/mars/core/graph/builder/chunk.py in _iter(self)
178 for tileable, tile_handler in \
179 self._gen_tileable_handlers(next_tileable_handlers):
--> 180 self._tile(chunk_graph, tileable, tile_handler,
181 next_tileable_handlers, to_update_tileables, visited)
182 self._tileable_handlers = next_tileable_handlers

~/Workspace/mars/mars/core/graph/builder/chunk.py in _tile(self, chunk_graph, tileable, tile_handler, next_tileable_handlers, to_update_tileables, visited)
92 visited: Set[EntityType]):
93 try:
---> 94 need_process = next(tile_handler)
95 if need_process is None:
96 chunks = []

~/Workspace/mars/mars/core/graph/builder/chunk.py in _tile_handler(self, tileable)
68 tiled_tileables = [self._get_data(t) for t in tiled_tileables]
69 # start to tile
---> 70 tiled_tileables = yield from handler.tile(tiled_tileables)
71 return tiled_tileables
72

~/Workspace/mars/mars/core/entity/tileables.py in tile(cls, tileables)
75 # they will be put into ChunkGraph and executed first.
76 # After execution, resume from the yield place.
---> 77 tiled_result = yield from tile_handler(op)
78 else:
79 # without iterative tiling

~/Workspace/mars/mars/dataframe/base/apply.py in tile(cls, op)
186 def tile(cls, op):
187 if op.inputs[0].ndim == 2:
--> 188 return (yield from cls._tile_df(op))
189 else:
190 return cls._tile_series(op)

~/Workspace/mars/mars/dataframe/base/apply.py in _tile_df(cls, op)
109 if axis == 1:
110 chunk_size = chunk_size[::-1]
--> 111 in_df = yield from recursive_tile(in_df.rechunk(chunk_size))
112
113 chunks = []

~/Workspace/mars/mars/core/entity/utils.py in recursive_tile(tileable, *tileables)
73 q.extend(cs)
74 continue
---> 75 yield from handler.tile(t.op.outputs)
76 q.pop()
77

~/Workspace/mars/mars/core/entity/tileables.py in tile(cls, tileables)
75 # they will be put into ChunkGraph and executed first.
76 # After execution, resume from the yield place.
---> 77 tiled_result = yield from tile_handler(op)
78 else:
79 # without iterative tiling

~/Workspace/mars/mars/dataframe/base/rechunk.py in tile(cls, op)
76 a = op.input
77 a = asdataframe(a) if a.ndim == 2 else asseries(a)
---> 78 chunk_size = _get_chunk_size(a, op.chunk_size)
79 if chunk_size == a.nsplits:
80 return [a]

~/Workspace/mars/mars/dataframe/base/rechunk.py in _get_chunk_size(a, chunk_size)
105 else:
106 itemsize = a.dtype.itemsize
--> 107 return get_nsplits(a, chunk_size, itemsize)
108
109

~/Workspace/mars/mars/tensor/rechunk/core.py in get_nsplits(tileable, new_chunk_size, itemsize)
36 chunk_size = new_chunk_size
37
---> 38 return decide_chunk_sizes(tileable.shape, chunk_size, itemsize)
39
40

~/Workspace/mars/mars/tensor/utils.py in decide_chunk_sizes(shape, chunk_size, itemsize)
560 raise ValueError("chunks have more dimensions than input tensor")
561 if nleft == 0:
--> 562 return normalize_chunk_sizes(shape, tuple(chunk_size[j] for j in range(len(shape))))
563
564 max_chunk_size = options.chunk_store_limit

~/Workspace/mars/mars/tensor/utils.py in normalize_chunk_sizes(shape, chunk_size)
68 chunk_sizes.append(chunk)
69 else:
---> 70 assert isinstance(chunk, int)
71
72 if size == 0:

AssertionError:
```

Start with mars/dataframe/base/apply.py, especially _tile_df, and follow the rechunk path shown in the traceback through mars/dataframe/base/rechunk.py and mars/tensor/utils.py. Run the provided filtered DataFrame and df.apply reproducer, then confirm that apply completes without the AssertionError when chunks on the apply axis are unknown.

Written by the indexing model from the issue text.