code: def read_ray_dataset(
      ds, columns=None, incremental_index=False, **kwargs
  ):
      assert isinstance(ds, real_ray_dataset.Dataset)
      refs = ds.to_pandas_refs()
      schema = ds.schema()
  
      import pyarrow as pa
  
      try:
          from ray.data._internal.pandas_block import PandasBlockSchema
      except ImportError:
          try:
              from ray.data.impl.pandas_block import PandasBlockSchema
          except ImportError:  # pragma: no cover
              PandasBlockSchema = type(None)
      try:
          from ray.data.dataset import Schema as RayDatasetSchema
      except ImportError:
          RayDatasetSchema = type(None)
  
      if isinstance(schema, PandasBlockSchema):
          dtypes = pd.Series(schema.types, index=schema.names)
      elif isinstance(schema, RayDatasetSchema):
          dtypes = pd.Series(
              [
                  t.to_pandas_dtype()
                  if t is not object
                  else np.dtype("O")
                  for t in schema.types
              ],
              index=schema.names,
          )
      elif isinstance(schema, pa.Schema):
          dtypes = schema.empty_table().to_pandas().dtypes
      else:
          raise NotImplementedError(
              f"Unsupported format of schema {schema}"
          )
  
      index_value = parse_index(pd.RangeIndex(-1))
      columns_value = parse_index(dtypes.index, store_data=True)
      op = DataFrameReadRayDataset(
          refs=refs,
          columns=columns,
          incremental_index=incremental_index,
      )
      return op(
          index_value=index_value,
          columns_value=columns_value,
          dtypes=dtypes,
      )
  
imports: import functools
  import warnings
  import numpy
  import pandas
  from ...None import opcodes
  from ...core import OutputType
  from ...serialization.serializables import AnyField, BoolField, ListField, Int64Field, ReferenceField
  from ..utils import parse_index, lazy_import, tokenize
  from .core import IncrementalIndexDatasource, IncrementalIndexDataSourceMixin, HeadOptimizedDataSource
label_desc: Use image node_id 1 for calling a global function with example usage: read_ray_dataset(ds, columns, incremental_index) and returns: op