Setting num_bytes to DEFAULT_READ_BUFFER_SIZE if it is None originally (#34475)

* Setting num_bytes to DEFAULT_READ_BUFFER_SIZE if it is None originally (resolves #34435)

* Setting num_bytes to DEFAULT_READ_BUFFER_SIZE if it is None originally (resolves #34435)

* Setting num_bytes to DEFAULT_READ_BUFFER_SIZE if it is None originally (resolves #34435)

* setting the default value in the function definition

Author: TanuSharma2511

sdks/python/apache_beam/io/filesystem.py

@@ -291,7 +291,7 @@ def _read_from_internal_buffer(self, read_fn):
     self._read_buffer.seek(0, os.SEEK_END)  # Allow future writes.
     return result
 
-  def read(self, num_bytes: Optional[int] = None) -> bytes:
+  def read(self, num_bytes: Optional[int] = DEFAULT_READ_BUFFER_SIZE) -> bytes:
     if not self._decompressor:
       raise ValueError('decompressor not initialized')
 

sdks/python/apache_beam/io/filesystem_test.py

@@ -470,6 +470,21 @@ def test_read_and_seek_back_to_beginning(self):
 
         self.assertEqual(first_pass, second_pass)
 
+  def test_read(self):
+    for compression_type in [CompressionTypes.BZIP2,
+                             CompressionTypes.DEFLATE,
+                             CompressionTypes.GZIP,
+                             CompressionTypes.ZSTD,
+                             CompressionTypes.LZMA]:
+      file_name = self._create_compressed_file(compression_type, self.content)
+      with open(file_name, 'rb') as f:
+        compressed_fd = CompressedFile(
+            f, compression_type, read_size=self.read_block_size)
+
+        data = compressed_fd.read()
+
+        self.assertEqual(data, self.content)
+
   def test_tell(self):
     lines = [b'line%d\n' % i for i in range(10)]
     tmpfile = self._create_temp_file()