rahil-c opened a new pull request, #19110:
URL: https://github.com/apache/hudi/pull/19110
### Describe the issue this Pull Request addresses
Today Hudi flattens the Debezium change-event envelope inside the Debezium
*sources* (`PostgresDebeziumSource` / `MysqlDebeziumSource`), which couples the
flattening logic to Kafka as the ingestion path. This PR extracts that logic
into standalone, source-agnostic `Transformer`s, so any source that produces
the raw Debezium envelope (Kafka, file/object-store CDC logs, etc.) can reuse
the same flattening.
### Summary and Changelog
Adds Debezium CDC transformers under
`org.apache.hudi.utilities.transform.debezium`:
- **`AbstractDebeziumTransformer`** — base transformer that flattens a
Debezium envelope (`{op, ts_ms, before, after, source}`) into a Hudi row. It
selects the `before` image for deletes and the `after` image otherwise and
explodes it to the top level, surfaces the common Debezium metadata columns
(operation type, upstream-processing / origin timestamps, shard), applies an
optional database-specific post-processing step, preserves the error-table
corrupt-record column when the error table is enabled, and normalizes column
nullability.
- **`PostgresDebeziumTransformer`** — surfaces the Postgres source metadata
(`txId`, `lsn`, `xmin`) as the flattened `_event_tx_id` / `_event_lsn` /
`_event_xmin` columns. Post-processing defaults a null `_event_lsn` to `0` for
snapshot rows (`op = "r"`), since the LSN is not populated for rows produced by
Debezium incremental snapshots and leaving it null would break LSN-based
ordering. Nests metadata by default.
- **`MysqlDebeziumTransformer`** — surfaces the MySQL binlog coordinates
(`file`, `pos`, `row`) as `_event_bin_file` / `_event_pos` / `_event_row`, and
derives the `_event_seq` ordering column as `"<binlog-file-suffix>.<pos>"`
(e.g. `"000001.100"` for binlog file `mysql-bin.000001` at position `100`).
Flat by default.
- **`DebeziumTransformerConfig`** — `hoodie.streamer.transformer.debezium.*`
configs:
- `nested.fields.enable` — groups the CDC metadata columns under a single
`_debezium_metadata` struct column instead of flattening them to the root
level. The operation-type column and the log-position column (e.g. the Postgres
LSN) stay at the root level so payload ordering keeps working. When unset, the
per-database transformer default applies (Postgres nests by default; MySQL is
flat by default).
- `schema.nullable.enable` — marks all columns in the transformed schema
nullable, matching the nullable columns Debezium change events produce.
Output column names match `DebeziumConstants`, so the existing
`PostgresDebeziumAvroPayload` and `MySqlDebeziumAvroPayload` merge/ordering
semantics keep working unchanged.
Nested-metadata-aware ordering inference:
- Adds the shared `DebeziumConstants.DEBEZIUM_METADATA_FIELD`
(`_debezium_metadata`) constant.
- `HoodieTableConfig.handlePayloadAdhocConfigs`: when nested metadata is
enabled, the MySQL ordering fields (`_event_bin_file`, `_event_pos`) move into
the `_debezium_metadata` struct, so the inferred ordering-field config now
references the nested path. The Postgres `_event_lsn` and the operation-type
column stay at the root level, so they are unchanged. A
`nestedDebeziumMetadataEnabled` flag is threaded through new overloads of
`inferMergingConfigsForWrites` / `inferMergingConfigsForV9TableCreation` (the
existing signatures are preserved and default to `false`); `HoodieStreamer`
supplies the real value from the config.
Tests: unit tests covering image selection (insert / update / delete), flat
vs. nested layout, the per-database nested default and its override, snapshot
LSN defaulting, MySQL `_event_seq` derivation, error-table corrupt-record
passthrough, the non-nullable-schema path, and nested vs. flat ordering-field
inference.
### Impact
Additive and opt-in. New transformers, two advanced configs, and
nested-aware ordering inference. The existing Debezium sources and payloads are
unchanged, and the new config-inference overloads default to the previous
behavior, so callers that do not pass the flag are unaffected.
### Risk Level
low — additive feature guarded by new, backward-compatible overloads
(existing signatures preserved, default `false`); covered by unit tests.
### Documentation Update
The new configs are documented inline via `withDocumentation()` on
`DebeziumTransformerConfig`. No website change required.
### Contributor's checklist
- [x] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
