use batch delete for GCS IO (#34835)

* use batch delete for GCS IO

* fixed the gcsio test

* fix format

* fixed format

* fixed the lint

Author: liferoad

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

@@ -265,10 +265,23 @@ def delete(self, path, recursive=False):
     bucket_name, blob_name = parse_gcs_path(path)
     bucket = self.client.bucket(bucket_name)
     if recursive:
-      # List and delete all blobs under the prefix.
-      blobs = bucket.list_blobs(prefix=blob_name)
-      for blob in blobs:
-        self._delete_blob(bucket, blob.name)
+      # List all blobs under the prefix.
+      blobs_to_delete = bucket.list_blobs(
+          prefix=blob_name, retry=self._storage_client_retry)
+      # Collect full paths for batch deletion.
+      paths_to_delete = [
+          f'gs://{bucket_name}/{blob.name}' for blob in blobs_to_delete
+      ]
+      if paths_to_delete:
+        # Delete them in batches.
+        results = self.delete_batch(paths_to_delete)
+        # Log any errors encountered during batch deletion.
+        errors = [f'{path}: {err}' for path, err in results if err is not None]
+        if errors:
+          _LOGGER.warning(
+              'Failed to delete some objects during recursive delete of %s: %s',
+              path,
+              ', '.join(errors))
     else:
       # Delete only the specific blob.
       self._delete_blob(bucket, blob_name)

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

@@ -75,8 +75,38 @@ def lookup_bucket(self, name):
     else:
       return self.create_bucket(name)
 
-  def batch(self):
-    pass
+  def batch(self, raise_exception=True):
+    # Return a mock object configured to act as a context manager
+    # and provide the necessary _responses attribute after __exit__.
+    # test_delete performs 3 deletions.
+    num_expected_responses = 3
+    mock_batch = mock.Mock()
+
+    # Configure the mock responses (assuming success for test_delete)
+    # These need to be available *after* the 'with' block finishes.
+    # We'll store them temporarily and assign in __exit__.
+    successful_responses = [
+        mock.Mock(status_code=204) for _ in range(num_expected_responses)
+    ]
+
+    # Define the exit logic
+    def mock_exit_logic(exc_type, exc_val, exc_tb):
+      # Assign responses to the mock instance itself
+      # so they are available after the 'with' block.
+      mock_batch._responses = successful_responses
+
+    # Configure the mock to behave like a context manager
+    mock_batch.configure_mock(
+        __enter__=mock.Mock(return_value=mock_batch),
+        __exit__=mock.Mock(side_effect=mock_exit_logic))
+
+    # The loop inside _batch_with_retry calls fn(request) for each item.
+    # The real batch object might have methods like add() or similar,
+    # but the core logic in gcsio.py calls the passed function `fn` directly
+    # within the `with` block. So, no specific action methods seem needed
+    # on the mock_batch itself for this test case.
+
+    return mock_batch
 
   def add_file(self, bucket, blob, contents):
     folder = self.lookup_bucket(bucket)