Skip to content

Make GCS filesystem lookup lazy to match S3 behavior #37488

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

Open
RushabhRatnaparkhi wants to merge 5 commits into
base: master
Choose a base branch
from
+105 −46
Open

Make GCS filesystem lookup lazy to match S3 behavior #37488

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0c43c2b
Fix GCSFileSystem to lazily load gcsio dependency
RushabhRatnaparkhi Jan 31, 2026
d7055ac
Make GCS filesystem lookup lazy
RushabhRatnaparkhi Feb 3, 2026
b4f6615
Fix Python formatting and lint issues
RushabhRatnaparkhi Feb 4, 2026
d0d5e84
Fix lazy GCS chunk size caching and lineage fallback without gcp deps
RushabhRatnaparkhi Feb 17, 2026
2d60f1f
Fix GCS filesystem test gating without gcp extras
RushabhRatnaparkhi Feb 21, 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
21 changes: 18 additions & 3 deletions sdks/python/apache_beam/io/gcp/gcsfilesystem.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,21 +34,34 @@
from apache_beam.io.filesystem import CompressionTypes
from apache_beam.io.filesystem import FileMetadata
from apache_beam.io.filesystem import FileSystem
from apache_beam.io.gcp import gcsio

__all__ = ['GCSFileSystem']


class GCSFileSystem(FileSystem):
"""A GCS ``FileSystem`` implementation for accessing files on GCS.
"""
def _get_gcsio(self):
try:
from apache_beam.io.gcp import gcsio
return gcsio
except ImportError:
raise ImportError(
'GCSFileSystem requires apache-beam[gcp]. '
'Install it with: pip install apache-beam[gcp]')

CHUNK_SIZE = gcsio.MAX_BATCH_OPERATION_SIZE # Chuck size in batch operations
GCS_PREFIX = 'gs://'

def __init__(self, pipeline_options):
super().__init__(pipeline_options)
self._pipeline_options = pipeline_options
self._chunk_size = None

@property
def CHUNK_SIZE(self):
if self._chunk_size is None:
self._chunk_size = self._get_gcsio().MAX_BATCH_OPERATION_SIZE
return self._chunk_size

@classmethod
def scheme(cls):
Expand Down Expand Up @@ -139,6 +152,7 @@ def _list(self, dir_or_prefix):
raise BeamIOError("List operation failed", {dir_or_prefix: e})

def _gcsIO(self):
gcsio = self._get_gcsio()
return gcsio.GcsIO(pipeline_options=self._pipeline_options)

def _path_open(
Expand Down Expand Up @@ -370,8 +384,9 @@ def delete(self, paths):

def report_lineage(self, path, lineage):
try:
gcsio = self._get_gcsio()
components = gcsio.parse_gcs_path(path, object_optional=True)
except ValueError:
except (ImportError, ValueError):
# report lineage is fail-safe
traceback.print_exc()
return
Expand Down
130 changes: 87 additions & 43 deletions sdks/python/apache_beam/io/gcp/gcsfilesystem_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@

from apache_beam.io.filesystem import BeamIOError
from apache_beam.io.filesystem import FileMetadata
from apache_beam.io.filesystems import FileSystems
from apache_beam.options.pipeline_options import PipelineOptions

# Protect against environments where apitools library is not available.
Expand All @@ -35,16 +36,31 @@
from apache_beam.io.gcp import gcsfilesystem
except ImportError:
gcsfilesystem = None # type: ignore

try:
from apache_beam.io.gcp import gcsio
except ImportError:
gcsio = None # type: ignore
# pylint: enable=wrong-import-order, wrong-import-position


@unittest.skipIf(gcsfilesystem is None, 'GCP dependencies are not installed')
class GCSFileSystemLazyLoadTest(unittest.TestCase):
def test_get_filesystem_does_not_require_gcp_extra(self):
fs = FileSystems.get_filesystem('gs://test-bucket/path')
self.assertEqual(fs.scheme(), 'gs')
Comment on lines +48 to +50
Copy link

Copilot AI Feb 4, 2026

Choose a reason for hiding this comment

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

The test validates that get_filesystem() succeeds without GCP extras, but it doesn't verify the second part of the lazy loading behavior: that actual filesystem operations (like open(), match(), etc.) properly raise an ImportError when GCP dependencies are missing. Consider adding a test that verifies an ImportError is raised when attempting to use the filesystem without GCP dependencies installed, to ensure the lazy loading works as intended.

Copilot uses AI. Check for mistakes.


@unittest.skipIf(
gcsfilesystem is None or gcsio is None,
'GCP dependencies are not installed')
class GCSFileSystemTest(unittest.TestCase):
def setUp(self):
assert gcsfilesystem is not None
pipeline_options = PipelineOptions()
self.fs = gcsfilesystem.GCSFileSystem(pipeline_options=pipeline_options)

def test_scheme(self):
assert gcsfilesystem is not None
self.assertEqual(self.fs.scheme(), 'gs')
self.assertEqual(gcsfilesystem.GCSFileSystem.scheme(), 'gs')

Expand Down Expand Up @@ -77,11 +93,13 @@ def test_split(self):
with self.assertRaises(ValueError):
self.fs.split('/no/gcs/prefix')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_match_multiples(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_match_multiples(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
gcsio_mock.list_files.return_value = iter([
('gs://bucket/file1', (1, 99999.0)),
('gs://bucket/file2', (2, 88888.0))
Expand All @@ -95,12 +113,14 @@ def test_match_multiples(self, mock_gcsio):
gcsio_mock.list_files.assert_called_once_with(
'gs://bucket/', with_metadata=True)

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_match_multiples_limit(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_match_multiples_limit(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
limit = 1
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
gcsio_mock.list_files.return_value = iter([
('gs://bucket/file1', (1, 99999.0))
])
Expand All @@ -111,11 +131,13 @@ def test_match_multiples_limit(self, mock_gcsio):
gcsio_mock.list_files.assert_called_once_with(
'gs://bucket/', with_metadata=True)

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_match_multiples_error(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_match_multiples_error(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
exception = IOError('Failed')
gcsio_mock.list_files.side_effect = exception

Expand All @@ -127,11 +149,13 @@ def test_match_multiples_error(self, mock_gcsio):
gcsio_mock.list_files.assert_called_once_with(
'gs://bucket/', with_metadata=True)

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_match_multiple_patterns(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_match_multiple_patterns(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
gcsio_mock.list_files.side_effect = [
iter([('gs://bucket/file1', (1, 99999.0))]),
iter([('gs://bucket/file2', (2, 88888.0))]),
Expand All @@ -141,33 +165,39 @@ def test_match_multiple_patterns(self, mock_gcsio):
result = self.fs.match(['gs://bucket/file1*', 'gs://bucket/file2*'])
self.assertEqual([mr.metadata_list for mr in result], expected_results)

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_create(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_create(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
# Issue file copy
_ = self.fs.create('gs://bucket/from1', 'application/octet-stream')

gcsio_mock.open.assert_called_once_with(
'gs://bucket/from1', 'wb', mime_type='application/octet-stream')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_open(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_open(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
# Issue file copy
_ = self.fs.open('gs://bucket/from1', 'application/octet-stream')

gcsio_mock.open.assert_called_once_with(
'gs://bucket/from1', 'rb', mime_type='application/octet-stream')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_copy_file(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_copy_file(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = ['gs://bucket/from1']
destinations = ['gs://bucket/to1']

Expand All @@ -177,11 +207,13 @@ def test_copy_file(self, mock_gcsio):
gcsio_mock.copy.assert_called_once_with(
'gs://bucket/from1', 'gs://bucket/to1')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_copy_file_error(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_copy_file_error(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = ['gs://bucket/from1']
destinations = ['gs://bucket/to1']

Expand All @@ -201,11 +233,13 @@ def test_copy_file_error(self, mock_gcsio):
gcsio_mock.copy.assert_called_once_with(
'gs://bucket/from1', 'gs://bucket/to1')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_copy_tree(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_copy_tree(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = ['gs://bucket1/']
destinations = ['gs://bucket2/']

Expand All @@ -215,11 +249,13 @@ def test_copy_tree(self, mock_gcsio):
gcsio_mock.copytree.assert_called_once_with(
'gs://bucket1/', 'gs://bucket2/')

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_rename(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_rename(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = [
'gs://bucket/from1',
'gs://bucket/from2',
Expand Down Expand Up @@ -255,11 +291,13 @@ def test_rename(self, mock_gcsio):
'gs://bucket/from3',
])

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_rename_delete_error(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_rename_delete_error(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = [
'gs://bucket/from1',
'gs://bucket/from2',
Expand Down Expand Up @@ -296,11 +334,13 @@ def test_rename_delete_error(self, mock_gcsio):
'gs://bucket/from3',
])

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_rename_copy_error(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_rename_copy_error(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
sources = [
'gs://bucket/from1',
'gs://bucket/from2',
Expand Down Expand Up @@ -335,11 +375,13 @@ def test_rename_copy_error(self, mock_gcsio):
'gs://bucket/from3',
])

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_delete(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_delete(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module
gcsio_mock._status.return_value = {'size': 0, 'updated': 99999.0}
files = [
'gs://bucket/from1',
Expand All @@ -351,11 +393,13 @@ def test_delete(self, mock_gcsio):
self.fs.delete(files)
gcsio_mock.delete_batch.assert_called()

@mock.patch('apache_beam.io.gcp.gcsfilesystem.gcsio')
def test_delete_error(self, mock_gcsio):
@mock.patch('apache_beam.io.gcp.gcsfilesystem.GCSFileSystem._get_gcsio')
def test_delete_error(self, mock_get_gcsio):
# Prepare mocks.
gcsio_mock = mock.MagicMock()
gcsfilesystem.gcsio.GcsIO = lambda pipeline_options=None: gcsio_mock
mock_gcsio_module = mock.MagicMock()
mock_gcsio_module.GcsIO.return_value = gcsio_mock
mock_get_gcsio.return_value = mock_gcsio_module

gcsio_mock._status.return_value = {'size': 0, 'updated': 99999.0}
files = [
Expand Down
Loading