[data] fix nested ragged ndarray

[raulchen]

Why are these changes needed?

Currently we support single level of ragged ndarray, i.e, shape of each row is different. But "nested ragged ndarray" isn't supported. I.e, when a row already contains a ragged ndarray, the following error will occur.

Repro:

import ray

def f(row):
    return {"result": [[], [1, 2]]}

ray.data.range(1).map(f).materialize()

Error:

ray.exceptions.RayTaskError(ArrowNotImplementedError): ray::Map(f)() (pid=39856, ip=127.0.0.1)
  File "/Users/chenh/code/ray/python/ray/data/_internal/execution/operators/map_operator.py", line 419, in _map_task
    for b_out in map_transformer.apply_transform(iter(blocks), ctx):
  File "/Users/chenh/code/ray/python/ray/data/_internal/execution/operators/map_transformer.py", line 393, in __call__
    add_fn(data)
  File "/Users/chenh/code/ray/python/ray/data/_internal/output_buffer.py", line 43, in add
    self._buffer.add(item)
  File "/Users/chenh/code/ray/python/ray/data/_internal/delegating_block_builder.py", line 24, in add
    check.build()
  File "/Users/chenh/code/ray/python/ray/data/_internal/table_block.py", line 128, in build
    tables = [self._table_from_pydict(columns)]
  File "/Users/chenh/code/ray/python/ray/data/_internal/arrow_block.py", line 143, in _table_from_pydict
    columns[col_name] = ArrowTensorArray.from_numpy(col, col_name)
  File "/Users/chenh/code/ray/python/ray/air/util/tensor_extensions/arrow.py", line 333, in from_numpy
    return ArrowVariableShapedTensorArray.from_numpy(arr)
  File "/Users/chenh/code/ray/python/ray/air/util/tensor_extensions/arrow.py", line 789, in from_numpy
    pa_dtype = pa.from_numpy_dtype(dtype)
  File "pyarrow/types.pxi", line 5140, in pyarrow.lib.from_numpy_dtype
  File "pyarrow/error.pxi", line 91, in pyarrow.lib.check_status
pyarrow.lib.ArrowNotImplementedError: Unsupported numpy type 17

Related issue number

Fixes #44235, #41078, #44062

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[bveeramani]

What's meant by "scalar list" here? I expect it to mean a list[float] like [1, 2, 3] and not a nested arrays like [[1], [2]], but with this implementation nested lists would be considered scalar lists.

[c21]

why we only need to check udf_return_col[0]?

[raulchen]

@bveeramani @c21 I've updated the PR with a new way to fix. I.e, do not convert each individual element, as long as the whole list is a nested list.

[bveeramani]

Don't think is_scalar_list is used anywhere other than here. Let's remove the function?