Skip to content

using fsspec adaptive-readahead cache #993

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
raj-prince wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

using fsspec adaptive-readahead cache #993

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
37b2e6d
using fsspec adaptive-readahead cache
raj-prince Aug 3, 2026
71d35a9
removing prefetcher with their test
raj-prince Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 15 additions & 48 deletions gcsfs/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
from .credentials import GoogleCredentials
from .inventory_report import InventoryReport
from .retry import errs, retry_request, validate_response
from .zb_hns_utils import DEFAULT_CONCURRENCY, MAX_PREFETCH_SIZE
from .zb_hns_utils import DEFAULT_CONCURRENCY

logger = logging.getLogger("gcsfs")

Expand Down Expand Up @@ -2049,7 +2049,7 @@ async def default_fetcher(start, size, split_factor=1):

fetcher_fn = default_fetcher

from .prefetcher import BackgroundPrefetcher
from fsspec.prefetcher import BackgroundPrefetcher

prefetcher = BackgroundPrefetcher(
fetcher=fetcher_fn,
Expand Down Expand Up @@ -2313,7 +2313,7 @@ def __init__(
mode="rb",
block_size=DEFAULT_BLOCK_SIZE,
autocommit=True,
cache_type="readahead",
cache_type="adaptive",
cache_options=None,
acl=None,
consistency="md5",
Expand Down Expand Up @@ -2376,6 +2376,18 @@ def __init__(
raise OSError("Attempt to open a bucket")
self.generation = _coalesce_generation(generation, path_generation)
self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY)

if cache_type is None:
cache_type = "adaptive"

if "r" in mode and cache_type == "adaptive":
if cache_type not in fsspec.core.caches:
warnings.warn(
"fsspec adaptive cache is unavailable in this environment; "
"falling back to readahead"
)
cache_type = "readahead"

super().__init__(
gcsfs,
path,
Expand All @@ -2394,34 +2406,6 @@ def __init__(
self.consistency = consistency
self.checker = get_consistency_checker(consistency)
Comment on lines 2406 to 2407

@gemini-code-assist gemini-code-assist Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Since self._prefetch_engine has been completely removed from GCSFile.__init__, calling _fetch_range on a ZonalFile instance will raise an AttributeError because ZonalFile._fetch_range (in gcsfs/zonal_file.py line 193) still accesses self._prefetch_engine directly. Defining self._prefetch_engine = None here maintains backward compatibility and prevents this runtime error.

Suggested change
self.consistency = consistency
self.checker = get_consistency_checker(consistency)
self.consistency = consistency
self.checker = get_consistency_checker(consistency)
self._prefetch_engine = None


# Ideally, all of these fields should be part of `cache_options`. Because current
# `fsspec` caches do not accept arbitrary `*args` and `**kwargs`, passing them
# there currently causes instantiation errors. We are holding off on introducing
# them as explicit keyword arguments to ensure existing user workloads are not
# disrupted. This will be refactored once the upstream `fsspec` changes are merged.
use_prefetch_reader = kwargs.get(
"use_experimental_adaptive_prefetching", False
) or os.environ.get(
"USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "false"
).lower() in (
"true",
"1",
)

if "r" in mode and use_prefetch_reader:
max_prefetch_size = kwargs.get("max_prefetch_size", MAX_PREFETCH_SIZE)
from .prefetcher import BackgroundPrefetcher

self._prefetch_engine = BackgroundPrefetcher(
self._async_fetch_range,
self.size,
max_prefetch_size=max_prefetch_size,
concurrency=self.concurrency,
loop=self.gcsfs.loop,
)
else:
self._prefetch_engine = None

# _supports_append is an internal argument not meant to be used directly.
# If True, allows opening file in append mode. This is generally not supported
# by GCS, but may be supported by subclasses (e.g. ZonalFile). This flag should
Expand Down Expand Up @@ -2612,8 +2596,6 @@ def _fetch_range(self, start=None, end=None):
if not both None, fetch only given range
"""
try:
if hasattr(self, "_prefetch_engine") and self._prefetch_engine:
return self._prefetch_engine.fetch(start=start, end=end)
return self.fs.cat_file(
self.path,
start=start,
Expand All @@ -2626,21 +2608,6 @@ def _fetch_range(self, start=None, end=None):
return b""
raise

async def _async_fetch_range(self, start_offset, total_size, split_factor=1):
"""Async fetcher mapped to the Prefetcher engine for regional buckets."""
return await self.gcsfs._cat_file_concurrent(
self.path,
start=start_offset,
end=start_offset + total_size,
concurrency=split_factor,
cache_type=self.cache_type,
)

def close(self):
super().close()
if hasattr(self, "_prefetch_engine") and self._prefetch_engine:
self._prefetch_engine.close()


def _convert_fixed_key_metadata(metadata, *, from_google=False):
"""
Expand Down
Loading