{"record":{"id":"74c8ce667d4ed0ae","repo":"apache/beam","slug":"compressor-not-initialized","errorCode":null,"errorMessage":"compressor not initialized","messagePattern":"compressor not initialized","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/filesystem.py","lineNumber":229,"sourceCode":"    elif self._compression_type == CompressionTypes.LZMA:\n      self._compressor = lzma.LZMACompressor()\n    else:\n      assert self._compression_type == CompressionTypes.GZIP\n      self._compressor = zlib.compressobj(\n          zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, self._gzip_mask)\n\n  def readable(self) -> bool:\n    mode = self._file.mode\n    return 'r' in mode or 'a' in mode\n\n  def writeable(self) -> bool:\n    mode = self._file.mode\n    return 'w' in mode or 'a' in mode\n\n  def write(self, data: bytes) -> None:\n    \"\"\"Write data to file.\"\"\"\n    if not self._compressor:\n      raise ValueError('compressor not initialized')\n    self._uncompressed_position += len(data)\n    compressed = self._compressor.compress(data)\n    if compressed:\n      self._file.write(compressed)\n\n  def _fetch_to_internal_buffer(self, num_bytes: int) -> None:\n    \"\"\"Fetch up to num_bytes into the internal buffer.\"\"\"\n    if (not self._read_eof and self._read_position > 0 and\n        (self._read_buffer.tell() - self._read_position) < num_bytes):\n      # There aren't enough number of bytes to accommodate a read, so we\n      # prepare for a possibly large read by clearing up all internal buffers\n      # but without dropping any previous held data.\n      self._read_buffer.seek(self._read_position)\n      data = self._read_buffer.read()\n      self._clear_read_buffer()\n      self._read_buffer.write(data)\n\n    assert self._decompressor","sourceCodeStart":211,"sourceCodeEnd":247,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/filesystem.py#L211-L247","documentation":"CompressedFile lazily creates its compressor only when opened for writing; write() requires that compressor to exist. Calling write on a CompressedFile that was constructed (or reopened) in read mode leaves self._compressor as None, so Beam raises ValueError.","triggerScenarios":"Opening a file in read mode (or via a path where writable() is False), then calling write(); using a CompressedFile after a mode reset that cleared the compressor.","commonSituations":"Opening with 'rb' but a code path still calls write(); a sink writing to a read-only file handle; accidentally swapping reader/writer objects.","solutions":["Open the underlying file (or FileSystems destination) in write mode ('wb'/'ab') so the compressor is initialized.","Check writable() (or mode) before writing and branch to a reader instead.","Fix the code path that hands a read-mode handle to a write API."],"exampleFix":"# before\nfh = open(path, 'rb')\nout = CompressedFile(fh, compression_type=CompressionTypes.GZIP)\nout.write(b'data')  # raises\n# after\nfh = open(path, 'wb')\nout = CompressedFile(fh, compression_type=CompressionTypes.GZIP)\nout.write(b'data')","handlingStrategy":"type-guard","validationCode":"if not compressed_file.writable():\n    raise ValueError('CompressedFile opened in read mode; cannot write')","typeGuard":"def can_write(f) -> bool:\n    return hasattr(f, 'writable') and f.writable()","tryCatchPattern":"try:\n    out.write(data)\nexcept ValueError as e:\n    if 'compressor not initialized' in str(e):\n        raise IOError('attempted write on read-mode CompressedFile') from e\n    raise","preventionTips":["Open write targets with 'wb'/'ab' modes only.","Keep reader and writer objects in clearly named variables to avoid mixups."],"tags":["python","apache-beam","io","invalid-state"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}