apache/beam · error · ValueError

decompressor not initialized

Error message

decompressor not initialized

What it means

CompressedFile lazily creates its decompressor only when opened for reading; read() requires it. Calling read on a CompressedFile that was opened for writing (or whose decompressor was never initialized) raises ValueError, since there is no codec to decode the stream.

Solutions

1. Open a separate CompressedFile in read mode ('rb') for reading instead of reusing the writer.
2. Guard reads with readable() and avoid seek() on write-mode handles.
3. Fix the mixed read/write code path — CompressedFile does not support dual-mode use.

Example fix

# before
fh = open(path, 'wb')
f = CompressedFile(fh, compression_type=CompressionTypes.GZIP)
f.seek(0)  # internally reads -> raises
# after
f = CompressedFile(open(path, 'rb'), compression_type=CompressionTypes.GZIP)
f.seek(0)

if not compressed_file.readable():
    raise ValueError('CompressedFile opened in write mode; cannot read')

def can_read(f) -> bool:
    return hasattr(f, 'readable') and f.readable()

try:
    data = f.read(n)
except ValueError as e:
    if 'decompressor not initialized' in str(e):
        raise IOError('attempted read on write-mode CompressedFile') from e
    raise

When it happens

Trigger: Opening a file in write mode ('wb') and then calling read(); invoking seek() on a write-mode CompressedFile, which internally calls read().

Common situations: Mixing read/write on the same CompressedFile handle; calling seek (documented to rely on read) on a writer; swapping reader/writer objects by mistake.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• compressor not initialized
• A schema is required to write non-schema'd data.
• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• An explicit schema is required to write non-schema'd…
• Cannot call read after iterating.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/bf1efbcc628c4542. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/io/filesystem.py:294

          # fully decompressing files.
          self._read_buffer.write(self._decompressor.flush())

        # Record that we have hit the end of file, so we won't unnecessarily
        # repeat the completeness verification step above.
        self._read_eof = True

  def _read_from_internal_buffer(self, read_fn):
    """Read from the internal buffer by using the supplied read_fn."""
    self._read_buffer.seek(self._read_position)
    result = read_fn()
    self._read_position += len(result)
    self._uncompressed_position += len(result)
    self._read_buffer.seek(0, os.SEEK_END)  # Allow future writes.
    return result

  def read(self, num_bytes: Optional[int] = DEFAULT_READ_BUFFER_SIZE) -> bytes:
    if not self._decompressor:
      raise ValueError('decompressor not initialized')

    self._fetch_to_internal_buffer(num_bytes)
    return self._read_from_internal_buffer(
        lambda: self._read_buffer.read(num_bytes))

  def readline(self) -> bytes:
    """Equivalent to standard file.readline(). Same return conventions apply."""
    if not self._decompressor:
      raise ValueError('decompressor not initialized')

    bytes_io = io.BytesIO()
    while True:
      # Ensure that the internal buffer has at least half the read_size. Going
      # with half the _read_size (as opposed to a full _read_size) to ensure
      # that actual fetches are more evenly spread out, as opposed to having 2
      # consecutive reads at the beginning of a read.
      self._fetch_to_internal_buffer(self._read_size // 2)
      line = self._read_from_internal_buffer(

View on GitHub (pinned to 12126d8942)