fix(declarative): detect gzip payloads in GzipParser instead of trusting headers

[Airbyte Support (Airbyte-Support)]

Summary

Requested by Syed Khadeer (Airbyte support) off Zendesk ticket 18622: a customer's Connector Builder Test read of a stream that downloads a gzipped CSV fails with UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8b in position 1 while a real sync of the same stream succeeds. 0x1f 0x8b is the gzip magic — compressed bytes are reaching the UTF-8 CSV parser.

Cause: create_gzip_decoder builds two different decoders depending on mode, and neither actually looks at the payload.

if self._emit_connector_builder_messages:      # Builder test read
    return CompositeRawDecoder(gzip_parser.inner_parser, False)   # gunzip step dropped entirely
return CompositeRawDecoder.by_headers(          # real sync
    [({"Content-Encoding", "Content-Type"}, _compressed_response_types, gzip_parser)],
    stream_response=True,
    fallback_parser=gzip_parser.inner_parser,   # no gunzip unless a header matched
)

The Builder branch assumed requests had already decompressed response.content, which only holds for transport-level Content-Encoding: gzip. When the body is a gzip payload (e.g. an S3 *.csv.gz download served as application/gzip or binary/octet-stream, as Apple App Store Connect analytics report segments are), nothing gunzips it in Builder mode. Measured on main with a single GzipDecoder(CsvDecoder):

payload / headers	sync	Builder test read
gzip, Content-Type: application/gzip	OK	UnicodeDecodeError 0x8b
gzip, Content-Type: binary/octet-stream	UnicodeDecodeError 0x8b	UnicodeDecodeError 0x8b
gzip, Content-Encoding: gzip	BadGzipFile	OK

That asymmetry is why manifests in the wild carry double-nested GzipDecoder(GzipDecoder(CsvDecoder)) workarounds (source-amazon-ads ships one next to a # TODO Fix me) — and why the single- vs. double-nesting workarounds are mutually exclusive depending on which header the server returns. Previously reported in airbytehq/oncall#7739, #11173, #11809 and airbytehq/airbyte#56988.

Fix: make GzipParser self-detecting — the behavior its own docstring already promised ("If the data is not gzipped, reset the pointer and pass the data to the inner parser as is") but never implemented — and then use it in both modes.

• GzipParser.parse reads the 2-byte header (looping, since response.raw.read(2) may short-read), gunzips when it is \x1f\x8b, and otherwise hands the inner parser the untouched stream. Because the stream may be non-seekable (response.raw in sync mode), the consumed header is restored by wrapping it in a private _PrefixedStream (io.RawIOBase) inside an io.BufferedReader, which keeps it usable by both gzip.GzipFile and CsvParser's TextIOWrapper.
• create_gzip_decoder now passes gzip_parser in the Builder branch and as the by_headers fallback_parser, so both modes behave identically regardless of Content-Type/Content-Encoding.

A single GzipDecoder now works for every combination in the table above, in both modes. Existing double-nested manifests keep working: the inner GzipParser sees already-decompressed data and passes it through instead of raising BadGzipFile.

Test plan

• New GzipParser unit tests in unit_tests/sources/declarative/decoders/test_composite_decoder.py: gzipped payload, non-gzip pass-through over both a seekable and a non-seekable stream, a stream that returns one byte per read(), empty payload, and nested GzipParser(GzipParser(CsvParser)) over a singly-gzipped body.
• New create_gzip_decoder tests in unit_tests/sources/declarative/parsers/test_model_to_component_factory.py covering emit_connector_builder_messages True and False against each header shape above — Builder mode had no coverage at all before.
• poetry run pytest unit_tests/sources/declarative/decoders/ unit_tests/sources/declarative/parsers/test_model_to_component_factory.py, poetry run ruff format --check . && poetry run ruff check ., poetry run mypy --config-file mypy.ini airbyte_cdk.

Link to Devin session: https://app.devin.ai/sessions/a1216bf2fe134b968ea1a3ffeb07d9aa
Requested by: Airbyte Support (@Airbyte-Support)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1787193372-gzip-decoder-passthrough#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1787193372-gzip-decoder-passthrough

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[Copilot]

Pull request overview

This PR fixes inconsistent gzip handling in the declarative GzipDecoder between Connector Builder “Test read” and real syncs by making GzipParser detect gzip content from the payload (magic bytes) instead of relying on response headers.

Changes:

• Update GzipParser to peek the first bytes of the stream, gunzip only when gzip magic bytes are present, otherwise pass the stream through unchanged (including support for non-seekable/short-read streams via a prefixed wrapper).
• Make create_gzip_decoder use GzipParser in both Builder and sync modes (including as the by_headers fallback parser) for consistent behavior.
• Add unit tests covering gzip payload detection, pass-through behavior, short-read streams, nested gzip parsers, and factory behavior across Builder vs sync modes.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py	Implements payload-based gzip detection in GzipParser and introduces a prefixed stream wrapper to support non-seekable streams.
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py	Switches gzip decoder construction to use GzipParser consistently in Builder and sync modes (including as fallback).
unit_tests/sources/declarative/decoders/test_composite_decoder.py	Adds focused GzipParser unit tests for gzip detection, short reads, non-seekable streams, empty payloads, and nested gzip parsers.
unit_tests/sources/declarative/parsers/test_model_to_component_factory.py	Adds factory-level tests ensuring create_gzip_decoder handles gzip payloads across header shapes and Builder vs sync modes.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[devin-ai-integration]

👍 On it. Valid — _PrefixedStream.__init__ should call super().__init__() so IOBase state (notably closed) is initialized before io.BufferedReader wraps it. Small and proportionate; adding it now and will follow up with the commit SHA.

[devin-ai-integration]

☑️ Resolved in 36b6431. super().__init__() is now the first statement of _PrefixedStream.__init__; unit tests, ruff and mypy all still pass.

[github-actions]

PyTest Results (Fast)

4 282 tests  +16   4 270 ✅ +16   8m 28s ⏱️ + 1m 28s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36b6431. ± Comparison against base commit db6f309.

[github-actions]

PyTest Results (Full)

4 285 tests  +16   4 273 ✅ +16   11m 9s ⏱️ +5s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36b6431. ± Comparison against base commit db6f309.