Newest updates break DaskOfflineStore with S3 parquets

Expected Behavior

In version 0.40.1 the Dask Offline store was able to read the data_source.path directly from the FileSource and retrieve the data from S3 using a path like: s3://<your-bucket>/<file-name>

Current Behavior

Failing to pull data because it is now appending the repo_path to the front of the s3 url.

Example:
/tmp/feast:s3//<your-bucket>/<file-name>

I believe this is because of a recent change: #4624 which is now not accepting the S3 url as a absolute Path

Steps to reproduce

Rebuilt my environment with latest tagged version 0.41.3
Reran my get_historical_features and call hung for a while then errored with the file path error not existing

Specifications

Version: 0.41.3
Platform: Linux
Subsystem: Debian

Possible Solution

Revert that change or allow a flag that would be able to bypass that breaking change
IF storage_options NOT None, Read parquet directly

@ntkathole

CC @ntkathole

@bjmccotter7192

@bjmccotter7192 Can you please confirm if latest version fixed this issue ?

Had the same issue:

driver_stats_source = FileSource(
    name="driver_hourly_stats_source",
    path="s3://feast-data/driver_stats.parquet",
    timestamp_field="event_timestamp",
    s3_endpoint_override="https://s3-basket/....ru",
    created_timestamp_column="created",
)

File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 476, in _filesystem_dataset
    fs, paths_or_selector = _ensure_single_source(source, filesystem)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 441, in _ensure_single_source
    raise FileNotFoundError(path)
FileNotFoundError: /tmp/tmpp4_p5yjf/s3:/feast-data/driver_stats.parquet

@makSSMZ

@makSSMZ
Can you please provide a longer stack trace ?

@ntkathole
I think you forgot to fix it also in feast.infra.offline_stores.file_source.FileSource.get_uri_for_file_path()
That might be the issue

@tmihalac

@tmihalac

.......:54260 - "POST /push HTTP/1.1" 500
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/parquet/core.py", line 1371, in __init__
    self._dataset = ds.dataset(path_or_paths, filesystem=filesystem,
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 794, in dataset
    return _filesystem_dataset(source, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 476, in _filesystem_dataset
    fs, paths_or_selector = _ensure_single_source(source, filesystem)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 441, in _ensure_single_source
    raise FileNotFoundError(path)
FileNotFoundError: /tmp/tmphakk65gr/s3:/feast-data/driver_stats.parquet
Traceback (most recent call last):
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/middleware/errors.py", line 165, in __call__
    await self.app(scope, receive, _send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 714, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 734, in app
    await route.handle(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 288, in handle
    await self.app(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 76, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 73, in app
    response = await f(request)
               ^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/fastapi/routing.py", line 301, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/fastapi/routing.py", line 212, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_server.py", line 315, in push
    store.push(**push_params)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_store.py", line 1485, in push
    self.write_to_offline_store(
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_store.py", line 1705, in write_to_offline_store
    provider.ingest_df_to_offline_store(feature_view, table)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/passthrough_provider.py", line 418, in ingest_df_to_offline_store
    self.offline_write_batch(self.repo_config, feature_view, table, None)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/passthrough_provider.py", line 219, in offline_write_batch
    self.offline_store.__class__.offline_write_batch(
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/offline_stores/dask.py", line 480, in offline_write_batch
    prev_table = pyarrow.parquet.read_table(
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/parquet/core.py", line 1793, in read_table
    dataset = ParquetDataset(
              ^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/parquet/core.py", line 1371, in __init__
    self._dataset = ds.dataset(path_or_paths, filesystem=filesystem,
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 794, in dataset
    return _filesystem_dataset(source, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 476, in _filesystem_dataset
    fs, paths_or_selector = _ensure_single_source(source, filesystem)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 441, in _ensure_single_source
    raise FileNotFoundError(path)
FileNotFoundError: /tmp/tmphakk65gr/s3:/feast-data/driver_stats.parquet
[2025-04-03 08:51:56 +0000] [23] [ERROR] Exception in ASGI application
Traceback (most recent call last):
  File "/opt/app-root/lib64/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 409, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/fastapi/applications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/applications.py", line 112, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/middleware/errors.py", line 187, in __call__
    raise exc
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/middleware/errors.py", line 165, in __call__
    await self.app(scope, receive, _send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 714, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 734, in app
    await route.handle(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 288, in handle
    await self.app(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 76, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/opt/app-root/lib64/python3.11/site-packages/starlette/routing.py", line 73, in app
    response = await f(request)
               ^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/fastapi/routing.py", line 301, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/fastapi/routing.py", line 212, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_server.py", line 315, in push
    store.push(**push_params)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_store.py", line 1485, in push
    self.write_to_offline_store(
  File "/opt/app-root/lib64/python3.11/site-packages/feast/feature_store.py", line 1705, in write_to_offline_store
    provider.ingest_df_to_offline_store(feature_view, table)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/passthrough_provider.py", line 418, in ingest_df_to_offline_store
    self.offline_write_batch(self.repo_config, feature_view, table, None)
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/passthrough_provider.py", line 219, in offline_write_batch
    self.offline_store.__class__.offline_write_batch(
  File "/opt/app-root/lib64/python3.11/site-packages/feast/infra/offline_stores/dask.py", line 480, in offline_write_batch
    prev_table = pyarrow.parquet.read_table(
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/parquet/core.py", line 1793, in read_table
    dataset = ParquetDataset(
              ^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/parquet/core.py", line 1371, in __init__
    self._dataset = ds.dataset(path_or_paths, filesystem=filesystem,
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 794, in dataset
    return _filesystem_dataset(source, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 476, in _filesystem_dataset
    fs, paths_or_selector = _ensure_single_source(source, filesystem)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/app-root/lib64/python3.11/site-packages/pyarrow/dataset.py", line 441, in _ensure_single_source
    raise FileNotFoundError(path)
FileNotFoundError: /tmp/tmphakk65gr/s3:/feast-data/driver_stats.parquet

@tmihalac

@tmihalac
We also tried to change the line locally just to test in the file lib/python3.11/site-packages/feast/infra/offline_stores/dask.py

if config.repo_path is not None and not Path(file_options.uri).is_absolute():
absolute_path = config.repo_path / file_options.uri
else:
absolute_path = Path(file_options.uri)

to

if config.repo_path is not None and not Path(file_options.uri).is_absolute():
absolute_path = file_options.uri
else:
absolute_path = Path(file_options.uri)

And the push worked. Maybe the problem is somewhere in that part.

#5208 will fix it

Thank you all for looking and working on this issue. I will follow both this issue and the #5208 for the final fix and then I'll update my environment and make sure everything is working as expected.

Navigation Menu

Repository navigation

Newest updates break DaskOfflineStore with S3 parquets #4753

Expected Behavior

Current Behavior

Steps to reproduce

Specifications

Possible Solution

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Newest updates break DaskOfflineStore with S3 parquets #4753

Description

Expected Behavior

Current Behavior

Steps to reproduce

Specifications

Possible Solution

Activity

franciscojavierarceo commented on Apr 2, 2025

ntkathole commented on Apr 2, 2025

makSSMZ commented on Apr 2, 2025

tmihalac commented on Apr 2, 2025

makSSMZ commented on Apr 3, 2025

makSSMZ commented on Apr 3, 2025

ntkathole commented on Apr 3, 2025

bjmccotter7192 commented on Apr 4, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions