Daft needs to be running on the Ray Runner for this operation

[snowch]

I'm trying to retrieve data from VAST Data Database to a Daft Dataframe:

import sys
import traceback
from typing import Any, List, Optional

import pyarrow as pa
import ray
import vastdb
from daft import context, from_ray_dataset
from ray import ObjectRef

if ray.is_initialized():
    ray.shutdown()

ray.init(
    address="ray://10.143.11.241:30001",
    runtime_env={"pip": ["daft", "vastdb", "pyarrow", "tensorflow", ]},
)

context.set_runner_ray(address="ray://10.143.11.241:30001", noop_if_initialized=True)


@ray.remote
def fetch_vast_dataset(
    endpoint: str,
    access_key: str,
    secret_key: str,
    bucket_name: str,
    schema_name: str,
    table_name: str,
    columns: Optional[list] = None,
    predicate: Optional[Any] = None,
    internal_row_id: Optional[bool] = None,
) -> from_ray_dataset:
    """
    Fetch data from VAST database as a Ray task
    """
    print(f"[FUNCTION] fetch_vast_dataset called for {table_name}", file=sys.stderr)

    try:
        try:
            session = vastdb.connect(
                endpoint=endpoint, access=access_key, secret=secret_key
            )
            print("[CONNECTION] VAST database connection established", file=sys.stderr)
        except Exception as conn_err:
            print(f"[ERROR] Connection failed: {conn_err}", file=sys.stderr)
            traceback.print_exc(file=sys.stderr)
            raise

        with session.transaction() as tx:
            bucket = tx.bucket(bucket_name)
            schema = bucket.schema(schema_name)
            table = schema.table(table_name)

            try:
                reader: pa.RecordBatchReader = table.select(
                    columns=columns, predicate=predicate, internal_row_id=internal_row_id
                )

                # distribute batches across the ray cluster
                arrow_refs = []
                for batch in reader:
                    pa_table = pa.Table.from_batches([batch])
                    arrow_refs.append(ray.put(pa_table))

                ray_dataset = ray.data.from_arrow_refs(arrow_refs)
                return from_ray_dataset(ray_dataset)

            except Exception as select_err:
                print(f"[ERROR] Data selection failed: {select_err}", file=sys.stderr)
                traceback.print_exc(file=sys.stderr)
                raise

    except Exception as e:
        print(f"[CRITICAL] Comprehensive error in fetch_vast_dataset: {e}", file=sys.stderr)
        traceback.print_exc(file=sys.stderr)
        raise ValueError(f"Error reading data from VAST: {e}")


def ray_data_task():
    daft_dataframe = ray.get(
        fetch_vast_dataset.remote(
            endpoint="http://172.200.201.1:80",
            access_key="...",
            secret_key="...",
            bucket_name="csnowdb",
            schema_name="social_media",
            table_name="tweets",
            columns=["id", "text"],
            # Uncomment the predicate if needed
            # predicate=(_.text.contains('AI')),
            internal_row_id=True,
        )
    )
    return daft_dataframe

daft_dataframe = ray_data_task()
print(daft_dataframe)

however, if I run this, I get:

Ray has already been initialized, Daft will reuse the existing Ray context and ignore the supplied address: 10.143.11.241:30001
(fetch_vast_dataset pid=757450, ip=10.1.171.5) [FUNCTION] fetch_vast_dataset called for tweets
(fetch_vast_dataset pid=757450, ip=10.1.171.5) [CONNECTION] VAST database connection established
(fetch_vast_dataset pid=757450, ip=10.1.171.5) [ERROR] Data selection failed: Daft needs to be running on the Ray Runner for this operation
(fetch_vast_dataset pid=757450, ip=10.1.171.5) Traceback (most recent call last):
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py", line 69](http://localhost:8888/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py#line=68), in fetch_vast_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py", line 39](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py#line=38), in _wrap
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return timed_func(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py", line 225](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py#line=224), in tracked_fn
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     result = fn(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py", line 165](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py#line=164), in from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return DataFrame._from_ray_dataset(ds)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3286](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py#line=3285), in _from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     raise ValueError("Daft needs to be running on the Ray Runner for this operation")
(fetch_vast_dataset pid=757450, ip=10.1.171.5) ValueError: Daft needs to be running on the Ray Runner for this operation
(fetch_vast_dataset pid=757450, ip=10.1.171.5) rolling back txid=0226d0000000053d due to:
(fetch_vast_dataset pid=757450, ip=10.1.171.5) Traceback (most recent call last):
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py", line 69](http://localhost:8888/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py#line=68), in fetch_vast_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py", line 39](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py#line=38), in _wrap
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return timed_func(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py", line 225](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py#line=224), in tracked_fn
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     result = fn(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py", line 165](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py#line=164), in from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return DataFrame._from_ray_dataset(ds)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3286](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py#line=3285), in _from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     raise ValueError("Daft needs to be running on the Ray Runner for this operation")
(fetch_vast_dataset pid=757450, ip=10.1.171.5) ValueError: Daft needs to be running on the Ray Runner for this operation
(fetch_vast_dataset pid=757450, ip=10.1.171.5) [CRITICAL] Comprehensive error in fetch_vast_dataset: Daft needs to be running on the Ray Runner for this operation
(fetch_vast_dataset pid=757450, ip=10.1.171.5) Traceback (most recent call last):
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py", line 69](http://localhost:8888/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py#line=68), in fetch_vast_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py", line 39](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py#line=38), in _wrap
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return timed_func(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py", line 225](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py#line=224), in tracked_fn
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     result = fn(*args, **kwargs)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py", line 165](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py#line=164), in from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     return DataFrame._from_ray_dataset(ds)
(fetch_vast_dataset pid=757450, ip=10.1.171.5)   File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3286](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py#line=3285), in _from_ray_dataset
(fetch_vast_dataset pid=757450, ip=10.1.171.5)     raise ValueError("Daft needs to be running on the Ray Runner for this operation")
(fetch_vast_dataset pid=757450, ip=10.1.171.5) ValueError: Daft needs to be running on the Ray Runner for this operation

---------------------------------------------------------------------------
RayTaskError(ValueError)                  Traceback (most recent call last)
Cell In[1], line 103
     99     return daft_dataframe
    102 # Execute the task and get the result
--> 103 daft_dataframe = ray_data_task()
    104 print(daft_dataframe)

Cell In[1], line 84, in ray_data_task()
     82 def ray_data_task():
     83     # Fetch data using the remote function
---> 84     daft_dataframe = ray.get(
     85         fetch_vast_dataset.remote(
     86             endpoint="http://172.200.201.1:80",
     87             access_key="...",
     88             secret_key="...",
     89             bucket_name="csnowdb",
     90             schema_name="social_media",
     91             table_name="tweets",
     92             columns=["id", "text"],
     93             # Uncomment the predicate if needed
     94             # predicate=(_.text.contains('AI')),
     95             internal_row_id=True,
     96         )
     97     )
     99     return daft_dataframe

File [/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/_private/auto_init_hook.py:21](http://localhost:8888/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/_private/auto_init_hook.py#line=20), in wrap_auto_init.<locals>.auto_init_wrapper(*args, **kwargs)
     18 @wraps(fn)
     19 def auto_init_wrapper(*args, **kwargs):
     20     auto_init_ray()
---> 21     return fn(*args, **kwargs)

File [/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/_private/client_mode_hook.py:102](http://localhost:8888/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/_private/client_mode_hook.py#line=101), in client_mode_hook.<locals>.wrapper(*args, **kwargs)
     98 if client_mode_should_convert():
     99     # Legacy code
    100     # we only convert init function if RAY_CLIENT_MODE=1
    101     if func.__name__ != "init" or is_client_mode_enabled_by_default:
--> 102         return getattr(ray, func.__name__)(*args, **kwargs)
    103 return func(*args, **kwargs)

File [/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/api.py:42](http://localhost:8888/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/api.py#line=41), in _ClientAPI.get(self, vals, timeout)
     35 def get(self, vals, *, timeout=None):
     36     """get is the hook stub passed on to replace `ray.get`
     37 
     38     Args:
     39         vals: [Client]ObjectRef or list of these refs to retrieve.
     40         timeout: Optional timeout in milliseconds
     41     """
---> 42     return self.worker.get(vals, timeout=timeout)

File [/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/worker.py:433](http://localhost:8888/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/worker.py#line=432), in Worker.get(self, vals, timeout)
    431     op_timeout = max_blocking_operation_time
    432 try:
--> 433     res = self._get(to_get, op_timeout)
    434     break
    435 except GetTimeoutError:

File [/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/worker.py:461](http://localhost:8888/opt/homebrew/Caskroom/miniconda/base/envs/raydaft39/lib/python3.9/site-packages/ray/util/client/worker.py#line=460), in Worker._get(self, ref, timeout)
    459         logger.exception("Failed to deserialize {}".format(chunk.error))
    460         raise
--> 461     raise err
    462 if chunk.total_size > OBJECT_TRANSFER_WARNING_SIZE and log_once(
    463     "client_object_transfer_size_warning"
    464 ):
    465     size_gb = chunk.total_size [/](http://localhost:8888/) 2**30

RayTaskError(ValueError): ray::fetch_vast_dataset() (pid=757450, ip=10.1.171.5)
    return timed_func(*args, **kwargs)
  File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py", line 225](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/analytics.py#line=224), in tracked_fn
    result = fn(*args, **kwargs)
  File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py", line 165](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/convert.py#line=164), in from_ray_dataset
    return DataFrame._from_ray_dataset(ds)
  File "[/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3286](http://localhost:8888/tmp/ray/session_2025-03-27_09-05-38_512045_11/runtime_resources/pip/51f3c9e3a92a2d64051a16d86067f8e46eecab31/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py#line=3285), in _from_ray_dataset
    raise ValueError("Daft needs to be running on the Ray Runner for this operation")
ValueError: Daft needs to be running on the Ray Runner for this operation

During handling of the above exception, another exception occurred:

ray::fetch_vast_dataset() (pid=757450, ip=10.1.171.5)
  File "[/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py", line 79](http://localhost:8888/var/folders/j6/195rqgcs37z88kmyck3fqg_h0000gn/T/ipykernel_86835/3165294760.py#line=78), in fetch_vast_dataset
ValueError: Error reading data from VAST: Daft needs to be running on the Ray Runner for this operation

Any suggestions what I am doing wrong?

Thanks!

[snowch]

Ok, I need to add the code daft.context.set_runner_ray(address=RAY_ADDRESS, noop_if_initialized=True) inside the function.

However, now I get:

TypeError: cannot pickle 'builtins.LogicalPlanBuilder' object

Can the daft dataframe be retreived using ray.get( )?

[colin-ho]

Answer on slack, but TLDR logical plans are currently not serializable. For example, right now they may contain non serializable objects like iceberg tables.

[ccmao1130]

Here's the conversation on Daft Slack: https://dist-data.slack.com/archives/C041NA2RBFD/p1743488181449099?thread_ts=1743248857.067229&cid=C041NA2RBFD