[Python] Remove per-open metadata RPC from BeamBlobReader

sdks/python/apache_beam/io/gcp/gcsio.py

@@ -705,6 +705,19 @@ def is_soft_delete_enabled(self, gcs_path):
 
 
 class BeamBlobReader(BlobReader):
+  """A reader for GCS blobs.
+
+  Note that constructing this reader does not issue any request to GCS. Object
+  metadata is fetched lazily by the underlying ``BlobReader``, on the first
+  read or seek.
+
+  Known limitation: doubly compressed objects, i.e. those stored with both
+  "content-encoding=gzip" and "content-type=application/gzip" (or
+  "application/x-gzip"), are not supported. Detecting this up front would
+  require an extra metadata request on every open, which is too costly to do
+  unconditionally. See
+  https://github.com/googleapis/google-cloud-python/issues/18423.
+  """
   def __init__(
       self,
       blob,
@@ -717,21 +730,6 @@ def __init__(
     # (https://cloud.google.com/storage/docs/transcoding).
     super().__init__(
         blob, chunk_size=chunk_size, retry=retry, raw_download=raw_download)
-    # TODO: Remove this after
-    # https://github.com/googleapis/python-storage/issues/1406 is fixed.
-    # As a workaround, we manually trigger a reload here. Otherwise, an internal
-    # call of reader.seek() will cause an exception if raw_download is set
-    # when initializing BlobReader(),
-    blob.reload()
-
-    # TODO: Currently there is a bug in GCS server side when a client requests
-    # a file with "content-encoding=gzip" and "content-type=application/gzip" or
-    # "content-type=application/x-gzip", which will lead to infinite loop.
-    # We skip the support of this type of files until the GCS bug is fixed.
-    # Internal bug id: 203845981.
-    if (blob.content_encoding == "gzip" and
-        blob.content_type in ["application/gzip", "application/x-gzip"]):
-      raise NotImplementedError("Doubly compressed files not supported.")
 
     self.enable_read_bucket_metric = enable_read_bucket_metric
     self.mode = "r"

sdks/python/apache_beam/io/gcp/gcsio_integration_test.py

@@ -281,9 +281,12 @@ class GcsIOReadGzipTest(unittest.TestCase):
   ]
 
   @parameterized.expand([
-      (gzip_test_files[0], CompressionTypes.UNCOMPRESSED, NotImplementedError),
-      (gzip_test_files[0], CompressionTypes.GZIP, NotImplementedError),
-      (gzip_test_files[0], CompressionTypes.AUTO, NotImplementedError),
+      # TODO(https://github.com/googleapis/google-cloud-python/issues/18423):
+      # Re-enable these once doubly compressed objects ("content-encoding=gzip"
+      # + "content-type=application/gzip") are handled again.
+      # (gzip_test_files[0], CompressionTypes.UNCOMPRESSED, NotImplementedError),
+      # (gzip_test_files[0], CompressionTypes.GZIP, NotImplementedError),
+      # (gzip_test_files[0], CompressionTypes.AUTO, NotImplementedError),
       (gzip_test_files[1], CompressionTypes.UNCOMPRESSED, UnicodeDecodeError),
       (gzip_test_files[1], CompressionTypes.GZIP, None),
       (gzip_test_files[1], CompressionTypes.AUTO, None),

sdks/python/setup.py

@@ -525,7 +525,7 @@ def get_portability_package_data():
               'google-auth-httplib2>=0.1.0,<0.3.0',
               'google-cloud-datastore>=2.0.0,<3',
               'google-cloud-pubsub>=2.1.0,<3',
-              'google-cloud-storage>=2.18.2,<4',
+              'google-cloud-storage>=3.0.0,<4',
               'google-cloud-resource-manager>=1.12.0,<2',
               'google-cloud-dataflow-client>=0.13.0,<0.14.0',
               # GCP packages required by tests