fix(connectors): Add unwrap_envelope transform and envelope detection to fix source-sink format mismatch by atharvalade · Pull Request #3197 · apache/iggy

atharvalade · 2026-04-29T16:48:12Z

Which issue does this PR close?

Rationale

Sources (e.g. Postgres) wrap row data in a DatabaseRecord envelope while sinks (e.g. Iceberg) expect flat JSON matching the target table schema — no shared contract exists, producing silent null failures.

What changed?

The Postgres source emits {table_name, operation_type, timestamp, data: {...}, old_data} envelopes, but the Iceberg sink's Arrow JSON reader maps these nested structures to top-level fields as null, silently violating non-nullable constraints.

This adds a reusable unwrap_envelope transform to the connector SDK that extracts a nested field (e.g. data) and promotes it as the top-level payload, plus explicit envelope detection in the Iceberg sink that errors with an actionable message instead of failing silently.

Local Execution

Passed
Pre-commit hooks ran (fmt, clippy, license-headers, trailing-whitespace, trailing-newline all clean; 119 tests pass across SDK + integration suites)

AI Usage

Opus 4.6
used for codebase exploration and following existing transform patterns
All 8 new unit tests pass locally, clippy/fmt clean, existing 111 tests unaffected
Yes, all code can be explained

codecov · 2026-04-29T17:22:36Z

Codecov Report

❌ Patch coverage is 93.47826% with 12 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.61%. Comparing base (2337883) to head (eaf9dc5).
⚠️ Report is 3 commits behind head on master.

Files with missing lines	Patch %	Lines
...nectors/sdk/src/transforms/json/unwrap_envelope.rs	96.27%	5 Missing and 1 partial ⚠️
core/connectors/sdk/src/transforms/mod.rs	0.00%	3 Missing ⚠️
...e/connectors/sdk/src/transforms/unwrap_envelope.rs	85.00%	3 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3197       +/-   ##
=============================================
- Coverage     74.18%   56.61%   -17.58%     
  Complexity      943      943               
=============================================
  Files          1200     1201        +1     
  Lines        109645    97136    -12509     
  Branches      86534    74025    -12509     
=============================================
- Hits          81345    54989    -26356     
- Misses        25553    39564    +14011     
+ Partials       2747     2583      -164

Components	Coverage Δ
Rust Core	`52.28% <93.47%> (-23.16%)`	⬇️
Java SDK	`58.44% <ø> (ø)`
C# SDK	`69.43% <ø> (ø)`
Python SDK	`81.43% <ø> (ø)`
Node SDK	`91.44% <ø> (ø)`
Go SDK	`39.91% <ø> (ø)`

Files with missing lines	Coverage Δ
core/connectors/sdk/src/transforms/json/mod.rs	`53.84% <ø> (ø)`
core/connectors/sdk/src/transforms/mod.rs	`26.66% <0.00%> (-2.97%)`	⬇️
...e/connectors/sdk/src/transforms/unwrap_envelope.rs	`85.00% <85.00%> (ø)`
...nectors/sdk/src/transforms/json/unwrap_envelope.rs	`96.27% <96.27%> (ø)`

... and 274 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

github-actions · 2026-05-07T00:53:50Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.

If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either @core on Discord or by mentioning them directly here on the PR.

Thank you for your contribution!

hubcio · 2026-05-14T12:11:21Z

@atharvalade please rebase this PR

/author

…nk format mismatch

atharvalade · 2026-05-17T06:32:54Z

@atharvalade please rebase this PR

/author

done

atharvalade · 2026-05-17T06:33:07Z

/ready

hubcio

review notes (not posted as line comments because targets are out-of-diff or already tracked):

the FFI (consume) return code is still discarded at core/connectors/runtime/src/sink.rs:642-650 (ConsumeCallback returns i32, sdk/src/sink.rs:28-36) and AutoCommit::When(AutoCommitWhen::PollingMessages) at core/connectors/runtime/src/sink.rs:467 commits offsets before the sink runs. this PR widens the blast radius: any envelope-detected batch now return Err(Error::InvalidRecord) from the sink, gets logged, and the offsets have already advanced. tracked in #2927 and #2928 but this PR adds a brand new fail path triggerable purely by source/sink misconfiguration, so they should land together.
core/connectors/sinks/delta_sink/src/sink.rs:106 has the same envelope-vs-flat-JSON mismatch and no detection or unwrap guidance. fixing one sink and leaving the other broken means #3174 is still reachable through the postgres → delta path. either replicate the detection (with the tighter shape suggested below) or, better, document unwrap_envelope as the single fix and drop sink-side sniffing entirely.
neither core/connectors/sdk/README.md nor core/connectors/sinks/iceberg_sink/README.md mention the new unwrap_envelope transform or the source-compatibility requirement for postgres_source → iceberg. discoverability is the whole point of this fix; please add at least a short entry to both.
architecturally, sink-side envelope sniffing is the wrong layer. transform discipline + docs is the real fix. if sink-side detection stays, it should be opt-in via config, not a hardcoded postgres-shaped heuristic.

overall, i'm not fan of this PR.

hubcio · 2026-05-20T18:08:18Z

+        return false;
+    };
+    obj.contains_key("table_name") && obj.contains_key("data")
+}


this heuristic checks 2 of the 5 keys on DatabaseRecord (table_name, operation_type, timestamp, data, old_data - see core/connectors/sources/postgres_source/src/lib.rs:110-116).

false positives: any legit iceberg table modeling audit logs, catalogs, or CDC metadata that happens to have table_name + data columns gets the whole batch rejected.

false negatives: Debezium / Kafka-Connect envelopes use before / after / op / source / payload and slip right through into JsonArrowReader, which then writes nulls - the exact bug #3174 is supposed to fix.

also it bakes a postgres-source shape into a generic iceberg sink. options: tighten to the full 4-or-5 key shape, or move detection behind an opt-in detect_envelope config flag, or drop sink-side sniffing entirely and rely on unwrap_envelope + docs.

Removed entirely, relying on transform and docs now

hubcio · 2026-05-20T18:08:18Z

        .collect();

+    if let Some(first) = msgs.first()
+        && looks_like_envelope(first)


the sniff only looks at msgs.first(). mixed batch where the first message is flat and the rest are envelopes (or vice versa) skips detection entirely - envelope rows then hit JsonArrowReader and write nulls, which is the silent-corruption mode #3174 was opened for. envelope-first drops the whole batch including any valid flat rows.

either scan all messages, or document and enforce a batch-homogeneity invariant somewhere upstream.

Dropped sink-side detection completely

hubcio · 2026-05-20T18:08:18Z

+             'unwrap_envelope' transform with field = \"data\" to your \
+             connector config to extract the inner payload."
+        );
+        return Err(Error::InvalidRecord);


Error::InvalidRecord is a unit variant; the actionable hint ("add an unwrap_envelope transform with field = data") lives only in the error! log and is lost the moment the caller receives the error. Error::InvalidRecordValue(String) already exists (sdk/src/lib.rs:389) and is used in delta_sink and influxdb_source - switch to it and carry the hint in the variant.

this is also a concrete instance of #3176 (overloaded InvalidRecord).

hubcio · 2026-05-20T18:08:18Z

+             (detected 'table_name' + 'data' fields). The Iceberg sink \
+             expects flat JSON matching the target table schema. Add an \
+             'unwrap_envelope' transform with field = \"data\" to your \
+             connector config to extract the inner payload."


two issues with this error message:

it hardcodes field = "data", which only matches the postgres envelope shape. if/when detection broadens to Debezium (after) or other shapes, the suggestion is wrong.

no connector id / plugin id in the log line, so in a multi-tenant deployment with several iceberg sinks you cannot tell which one rejected the batch.

hubcio · 2026-05-20T18:08:18Z

+/// entire payload with the contents of `data`.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UnwrapEnvelopeConfig {
+    pub field: String,


UnwrapEnvelopeConfig.field: String has no validation. an empty string deserializes fine through from_config at transforms/mod.rs:137-141, then every message hits the missing-field branch and warn-spams at message rate (compounds the log flood on the json side).

reject empty field either via a custom Deserialize impl, or inline in UnwrapEnvelope::new, or in the from_config arm.

new() now rejects empty field with InvalidConfigValue

hubcio · 2026-05-20T18:08:18Z

        }
+        TransformType::UnwrapEnvelope => {
+            let cfg: UnwrapEnvelopeConfig =
+                serde_json::from_value(raw.clone()).map_err(|_| Error::InvalidConfig)?;


.map_err(|_| Error::InvalidConfig) throws away the serde error detail. for a user with a typo in field or a wrong type, all they see is InvalidConfig - no line number, no field name, no expected-vs-got. this is a pre-existing pattern in every arm of from_config, but this PR adds another instance.

worth either fixing all arms in a follow-up, or at least propagating the serde error through Error::InvalidConfigDetail(String) or similar.

Now propagates detail via InvalidConfigValue(format!(...))

…bility

atharvalade · 2026-05-22T05:51:29Z

removed all sink-side sniffing and shifted to transform discipline + docs (SDK, Iceberg, and Postgres READMEs updated). The FFI return code and AutoCommit issues are pre-existing (#2927/#2928) and out of scope now that this PR no longer adds a new fail path through that code @hubcio

atharvalade · 2026-05-22T05:51:50Z

/ready

hubcio · 2026-05-22T11:51:47Z

+field = "data"
+```
+
+Alternatively, configure the Postgres source with `payload_column` and `payload_format = "json_direct"` to bypass the envelope entirely — see the Postgres source README for details.


come on, don't leave postgres reference in iceberg readme. go thru all changes in this PR and make sure that all docs (readmes, comments, etc) are aligned with the actual code and placement in the project.

atharvalade changed the title ~~Add unwrap_envelope transform and envelope detection to fix source-sink format mismatch~~ fix(connectors): Add unwrap_envelope transform and envelope detection to fix source-sink format mismatch Apr 29, 2026

github-actions Bot added S-stale Inactive issue or pull request and removed S-stale Inactive issue or pull request labels May 7, 2026

github-actions Bot added the S-waiting-on-author PR is waiting on author response label May 14, 2026

atharvalade added 2 commits May 17, 2026 01:31

Add unwrap_envelope transform and envelope detection to fix source-si…

5c7e500

…nk format mismatch

Trigger CI

7c92361

atharvalade force-pushed the fix/shared-source-sink-contract branch from 8defa26 to 7c92361 Compare May 17, 2026 06:32

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels May 17, 2026

hubcio requested changes May 20, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels May 20, 2026

atharvalade and others added 2 commits May 21, 2026 22:45

Add unwrap_envelope transform and docs for source-sink format compati…

2b299ab

…bility

Merge branch 'master' into fix/shared-source-sink-contract

eaf9dc5

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels May 22, 2026

hubcio requested changes May 22, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels May 22, 2026

Conversation

atharvalade commented Apr 29, 2026

Which issue does this PR close?

Rationale

What changed?

Local Execution

AI Usage

Uh oh!

codecov Bot commented Apr 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions Bot commented May 7, 2026

Uh oh!

hubcio commented May 14, 2026

Uh oh!

atharvalade commented May 17, 2026

Uh oh!

atharvalade commented May 17, 2026

Uh oh!

hubcio left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

atharvalade commented May 22, 2026

Uh oh!

atharvalade commented May 22, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented Apr 29, 2026 •

edited

Loading