feat(utilities): add Postgres and Mysql Debezium CDC transformers

[rahil-c]

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 Transformers, 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

• Read through contributor's guide
• Enough context is provided in the sections above
• Adequate tests were added if applicable

[hudi-agent]

⚠️ 🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! This PR extracts the Debezium change-event flattening out of the Postgres/Mysql source classes into standalone transformers and adds an optional nested-metadata layout, with good test coverage. The main thing worth double-checking is whether the flat path should still surface db_schema_source_partition for parity with the existing PostgresDebeziumSource, plus a couple of smaller robustness notes in the inline comments. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A few naming and simplification suggestions below.

[hudi-agent]

🤖 In flat mode the source.schema → db_schema_source_partition column isn't surfaced, but the original PostgresDebeziumSource.processDataset always emitted it at root and the nested branch here still adds it (lines 154-155). So a Postgres pipeline adopting this transformer with nested.fields.enable=false would silently lose db_schema_source_partition — which the *_source_partition naming suggests is often a partition field. Is dropping it in flat mode intended, or should hasSchemaField be honored here too?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[wombatu-kun]

Confirmed: PostgresDebeziumSource.processDataset emitted db_schema_source_partition at root for both images, so flat mode here is a parity drop. Two caveats worth weighing: db_schema_source_partition is not in DebeziumConstants.META_COLUMNS and has no in-repo consumer, so the partition-field impact is limited to external pipelines that key on it; and the converse also holds - because Postgres nests by default, db_shard_source_partition (which IS in META_COLUMNS) moves off root too under the default config. If parity is the goal, honor hasSchemaField in the flat branch; otherwise call out the dropped/relocated columns in the config docs.

[rahil-c]

Not applicable for this PR — this is intentional/pre-existing behavior (the schema column is DB-dependent and only surfaced when nesting is enabled) and out of scope for this change.

[hudi-agent]

🤖 The legacy generateUniqueSequence UDF validated the binlog coords (fileId == null || trim().isEmpty() || pos == null || pos < 0) and threw a descriptive HoodieReadFromSourceException before building the seq. Here a null file/pos would produce a null _event_seq, which only fails later (and less clearly) in MySqlDebeziumAvroPayload.shouldPickCurrentRecord. Since _event_seq is the ordering/precombine field, might be worth keeping an explicit guard so malformed events fail fast at transform time.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[wombatu-kun]

Confirmed: the original MysqlDebeziumSource.generateUniqueSequence threw HoodieReadFromSourceException on file == null || file.trim().isEmpty() || pos == null || pos < 0 before building the key. applySeqNo drops that guard, and since Spark concat returns null when any argument is null, a malformed event yields a null _event_seq - the precombine field - which only surfaces later in MySqlDebeziumAvroPayload. Restoring an equivalent fail-fast check (or a coalesce plus assertion) in applySeqNo keeps the early, descriptive failure.

[rahil-c]

Not applicable for this PR — this is pre-existing, intentional behavior and out of scope for this change. Open to adding an explicit guard as a follow-up if there's appetite for it.

[hudi-agent]

🤖 Worth adding that the deferred failure is asymmetric, which strengthens the fail-fast case. In MySqlDebeziumAvroPayload.shouldPickCurrentRecord, a null _event_seq on the incoming record throws HoodieDebeziumAvroPayloadException, but a null seq on the stored record is swallowed by the bootstrap branch (!currentSourceSeqOpt.isPresent() → return false), so that record gets overwritten regardless of ordering rather than erroring. And a first-write of a malformed event goes through the insert path, so it can land silently and only trip later; preCombine would then hit Long.parseLong("null") → NumberFormatException. A guard in applySeqNo avoids all three of those less-obvious paths.

[hudi-agent]

🤖 Identifying the LSN column via col.toString().contains("_event_lsn") is a little fragile — it depends on how Spark renders the alias in Column.toString() and would also match any future type-specific column whose expression text happens to contain _event_lsn. Since the subclass already builds typeSpecificMetadataColumns and knows which one is the LSN, could it pass that explicitly (e.g. a dedicated constructor param) instead of string-matching?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[wombatu-kun]

This works today only because Spark's Alias.toString() embeds the alias name (it renders as source.lsn AS _event_lsn); any change to how the column is constructed or rendered would silently break LSN-to-root routing with no test catching it. Passing the LSN column (or its output name) explicitly from the subclass, as suggested, removes the string match entirely.

[rahil-c]

Not applicable for this PR — this is pre-existing behavior and out of scope for this change.

[hudi-agent]

🤖 nit: DEFAULT_NESTED_METADATA_COLUMNS is a bit misleading — in flat mode these columns are spread directly to root level (line 169), so calling them "nested" implies they only live inside the struct. Something like DEFAULT_COMMON_METADATA_COLUMNS would better capture that their placement varies by mode.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[rahil-c]

Not applicable for this PR — the naming is pre-existing and out of scope for this change.

[hudi-agent]

🤖 nit: Arrays.stream(...).collect(Collectors.toList()).contains(...) could be simplified to Arrays.asList(rowDataset.columns()).contains(...) — same result with less noise.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[rahil-c]

Not applicable for this PR — this is pre-existing style and out of scope for this change.

[hudi-agent]

🤖 nit: useDefaultValuesForLsnIfNull reads more like a boolean flag or predicate ("should we use defaults?") than a transformation action. Something like defaultNullSnapshotLsn or fillMissingSnapshotLsn would more clearly signal that this method mutates the dataset.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[rahil-c]

Not applicable for this PR — pre-existing naming and out of scope for this change.

[wombatu-kun]

This class has only protected constructors, no abstract methods, and a javadoc stating it is meant to be subclassed, yet it is declared as a concrete public class. Every other Abstract* type in the repo is declared abstract (e.g. its sibling AbstractDebeziumAvroPayload), and nothing instantiates this one directly. Declare it public abstract class so the name matches the contract and direct instantiation is prevented at compile time.

[rahil-c]

@rmahindra @yihua let me know if we need any integration test or heavier functional test or if the unit tests in this pr suffice?

[nsivabalan]

Thanks for upstreaming this, Rahil! I cross-checked the transformer core against a battle-tested internal implementation — before/after image selection, nested vs. flat metadata assembly, Postgres LSN-default-for-snapshot, MySQL _event_seq derivation, the schema.nullable.enable=false rebuild, and the error-table passthrough all line up. The nullability fix commits at the end of the history got that pass looking clean.

A few things worth addressing before merge:

---

🚨 The nestedDebeziumMetadataEnabled boolean on HoodieTableConfig shouldn't exist

HoodieTableConfig.handlePayloadAdhocConfigs already unconditionally overwrites the caller's ordering field for Debezium payloads (that's the pre-existing lines 932 and 936). This PR extends that overwrite to also know about the nested layout by threading a nestedDebeziumMetadataEnabled flag through two new overloads. That leaks Debezium layout semantics into hudi-common and forces every caller of inferMergingConfigsForWrites / inferMergingConfigsForV9TableCreation (reader path in HoodieReaderContext, HoodieSparkSqlWriter, MergeIntoHoodieTableCommand x2, Flink StreamerUtil, HoodieTableMetaClient init) to either pass the flag or silently default it to false.

Only HoodieStreamer was updated to pass the real flag. The rest silently default to false, which is fine for the reader (hoodie.properties already has it) but silently wrong for the SparkSQL/MetaClient auto-create paths if a user ever writes Debezium data with nested metadata through them.

Cleaner cut: remove the ORDERING_FIELDS assignment for Debezium payloads from handlePayloadAdhocConfigs entirely, and have the caller pass the correct ordering field name via the existing orderingFieldName argument. HoodieStreamer is the one place that knows the transformer is in the pipeline and whether nesting is on, so it should compute:

// In HoodieStreamer (or a small helper next to DebeziumTransformerConfig):
private static String resolveDebeziumOrderingFields(String payloadClass, String userOrdering, TypedProperties props) {
  boolean nested = ConfigUtils.getBooleanWithAltKeys(props, DebeziumTransformerConfig.ENABLE_NESTED_FIELDS);
  if (PostgresDebeziumAvroPayload.class.getName().equals(payloadClass)) {
    return DebeziumConstants.FLATTENED_LSN_COL_NAME; // stays at root even when nested
  }
  if (MySqlDebeziumAvroPayload.class.getName().equals(payloadClass)) {
    String prefix = nested ? DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." : "";
    return prefix + DebeziumConstants.FLATTENED_FILE_COL_NAME + "," + prefix + DebeziumConstants.FLATTENED_POS_COL_NAME;
  }
  return userOrdering;
}

and pass the result into the existing 5-arg inferMergingConfigsForWrites. No new overload, no boolean flag, no maybeNestColumn, hudi-common stays layout-agnostic, and the "silently wrong for other write paths" problem disappears because there's no boolean to silently default to false. TestHoodieTableConfigDebeziumOrdering can be dropped or shrunk to a one-liner that verifies the ordering field passes through unchanged; the real "does nesting produce the right ordering field?" test moves next to the transformer/streamer where the behavior lives.

One sanity check when doing this: PostgresDebeziumSource / MysqlDebeziumSource (the pre-existing non-transformer path) also flow through HoodieStreamer, so the resolver keying off payload class means those users pick up the same ordering-field resolution automatically — no regression, just consistent behavior.

---

⚠️ Postgres nested-metadata default should be flipped to false

PostgresDebeziumTransformer:61 currently passes nestedFieldsEnabledByDefault=true, so Postgres pipelines land in the nested layout unless the user explicitly opts out. MySQL defaults to false (flat). Two reasons to flip Postgres to false:

1. Cross-source consistency — both transformers should behave the same by default; users switching between Postgres and MySQL shouldn't get a different metadata layout for free.
2. Backward compatibility — existing pipelines built around the flat layout (what today's PostgresDebeziumSource produces) keep working with no config change.

Once flipped, delete the "PostgresDebeziumTransformer defaults to true" sentence from DebeziumTransformerConfig.java:57-58 and from the Javadoc at PostgresDebeziumTransformer.java:46-48.

---

⚠️ Visibility of new HoodieTableConfig.inferMergingConfigsFor… overloads

If 🚨 above is addressed the new overloads go away, and this is moot. If not, mark them @ApiStatus.Internal (or move to package-private) — they're public static today with no visibility marker.

---

💬 Suggestions

• Extract SNAPSHOT_OP = "r" into DebeziumConstants. DELETE_OP = "d" already lives centrally; the local constant in PostgresDebeziumTransformer.java:53 should join it.

• Class Javadoc on AbstractDebeziumTransformer should reference DebeziumTransformerConfig so operators land in the right place when they need to configure the transformer.

• Refactor AbstractDebeziumTransformer.apply() into named phases. It's ~95 lines and mixes three phases that the class-level Javadoc already lists as separate responsibilities. Extract them so the top-level apply() reads like the doc:

  @Override
  public Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, Dataset<Row> rowDataset, TypedProperties props) {
    if (rowDataset.columns().length == 0) {
      return rowDataset;
    }
    Dataset<Row> withDataField = selectBeforeOrAfterImage(rowDataset);
    List<Column> outputColumns = buildOutputColumns(withDataField, props);
    Dataset<Row> withErrorCol = applyErrorTablePassthrough(withDataField, outputColumns, props);
    Dataset<Row> flattened = withErrorCol.select(outputColumns.toArray(new Column[]{}));
    Dataset<Row> postProcessed = postProcessingOption.map(fn -> fn.apply(flattened)).orElse(flattened);
    return applyNullabilityRules(spark, rowDataset, postProcessed, props);
  }

  Where the extracted phases are:

  • selectBeforeOrAfterImage — the .withColumn(DATA_FIELD, when(op==d, before).otherwise(after)) bit.
  • buildOutputColumns — the flat-vs-nested branch that assembles allColumns and, in the nested case, adds the _debezium_metadata struct.
  • applyErrorTablePassthrough — the error-table corrupt-record handling.
  • applyNullabilityRules — merges both nullability branches (this is the phase most worth naming, since two commits in the history are about getting it right).

  Two notes: (a) the interior rowDataset = reassignments in the current code mean phases share mutable state through the outer variable — extraction forces threading the dataset through method returns instead, which is what makes the phases actually independent; (b) don't add abstract methods or subclass extension points while doing this — the class Javadoc explicitly says "there is no abstract method to implement," keep it that way. Purely private-method extraction.

• hasSchemaField helper could read cleaner (AbstractDebeziumTransformer.java:233-243). Not strictly wrong, but the double-nested Arrays.stream(...).filter(...).findFirst().map(source -> Arrays.stream(...).anyMatch(...)) chain packs two lookups into one expression. A small extraction:

  private static boolean hasSchemaField(Dataset<Row> rowDataset) {
    return getSourceStruct(rowDataset)
        .map(source -> Arrays.stream(source.fields()).anyMatch(f -> "schema".equals(f.name())))
        .orElse(false);
  }
  
  private static Optional<StructType> getSourceStruct(Dataset<Row> rowDataset) {
    return Arrays.stream(rowDataset.schema().fields())
        .filter(f -> DebeziumConstants.INCOMING_SOURCE_FIELD.equals(f.name()) && f.dataType() instanceof StructType)
        .map(f -> (StructType) f.dataType())
        .findFirst();
  }

  Or, if there's an existing "look up nested struct by name" helper elsewhere in the codebase, use that instead.

• Config resolution model changed shape vs. how it's typically done. Internal-style Debezium transformers read a single global flag inside apply(); the PR moves the default into the subclass constructor and OR's it with the config lookup. Once Postgres flips to false (⚠️ above) this becomes a non-issue behaviorally, but worth a one-line Javadoc note on the constructor explaining "subclass default; overridable via hoodie.streamer.transformer.debezium.nested.fields.enable" so future readers understand the resolution order.

• LSN column detection via col.toString() in AbstractDebeziumTransformer.java:139-146. Column.toString() isn't part of Spark's stable API — it returns things like "source.lsn AS _event_lsn", and the contains(FLATTENED_LSN_COL_NAME) works today only because both the alias and the source expression happen to include that substring. No existing precedent in the Hudi codebase to be consistent with either way, so happy to leave it — but if the refactor above lands where subclasses declare their "keep at root when nested" column explicitly, the string-matching loop goes away for free.

• Are Oracle/SQLServer transformers + type-conversion helpers planned follow-up patches? Please confirm in the PR body so reviewers and downstream users know what's coming.

---

💅 Nits

• AbstractDebeziumTransformer.java:47 — the imported symbol ERROR_TABLE_CURRUPT_RECORD_COL_NAME has a typo (CURRUPT → CORRUPT) inherited from BaseErrorTableWriter. Not this PR's problem to fix but worth flagging in a separate cleanup pass.
• PostgresDebeziumTransformer.java:44 — Javadoc link uses http:// not https://. Trivial.
• HoodieTableConfig.java:906 — stray .s in the comment "Additional custom merge properties.s". Pre-existing typo, easy drive-by fix if you're already touching adjacent lines. (Moot if 🚨 above removes this block entirely.)

---

Suggested tests (for after the refactor above lands)

1. End-to-end ORDERING_FIELDS resolution test in HoodieStreamer with nested metadata enabled — verify that with hoodie.streamer.transformer.debezium.nested.fields.enable=true and MySQL payload, the resulting hoodie.properties on disk has hoodie.record.merge.ordering.fields=_debezium_metadata._event_bin_file,_debezium_metadata._event_pos. Nothing currently proves the end-to-end wiring.
2. Postgres snapshot LSN defaulting under nested layout — testNullLsnDefaultedToZeroForSnapshotRows forces flat. Add a case with the default (ENABLE_NESTED_FIELDS unset) once Postgres nested-default is resolved either way.
3. MySQL applySeqNo when file or pos is null — no coverage today for null binlog coordinates. Whether the null-→-null propagation is correct is a design call; make it explicit either way since _event_seq is the payload's ordering field and a null there would break merge.

[rahil-c]

Thanks for the thorough review, @nsivabalan! Summary of the changes addressing the feedback (latest commit 58f39b5).

🚨 nestedDebeziumMetadataEnabled boolean / new HoodieTableConfig overloads

Removed. There's no boolean and no new overloads — inferMergingConfigsForWrites / inferMergingConfigsForV9TableCreation are back to their original signatures (unchanged vs. master) and maybeNestColumn is gone.

One deviation from the literal suggestion (resolve entirely in HoodieStreamer) that I'd like your read on: not every Debezium write path goes through the streamer. The Spark DataFrame/SQL and Flink table-creation paths reach HoodieTableConfig through HoodieTableMetaClient.init directly and rely on handlePayloadAdhocConfigs auto-correcting the ordering field to the payload's canonical column(s) — TestPayloadDeprecationFlow#testMergerBuiltinPayloadFromTableCreationPath exercises exactly that create path, and merge correctness depends on it. Resolving only in HoodieStreamer regresses those paths.

So instead of a layout flag, handlePayloadAdhocConfigs now reconciles using the existing orderingFieldName argument it already receives: it forces the flat canonical ordering (_event_lsn, or _event_bin_file,_event_pos) unless the caller already resolved the nested form — which only the Streamer does — in which case the nested form is preserved. The column knowledge lives in DebeziumConstants.resolveOrderingFields(payloadClass, nested), so the generic inference stays layout-agnostic. HoodieStreamer resolves the (flat or nested) ordering into cfg.sourceOrderingFields, so a nested table persists the nested path at creation. This also fixes a gap where the nested ordering field wasn't actually reaching table creation. Happy to iterate if you'd prefer a different split.

⚠️ Postgres nested-metadata default → false

Flipped for cross-source consistency with MySQL + backward compatibility. Removed the "defaults to true" wording from DebeziumTransformerConfig and the PostgresDebeziumTransformer Javadoc.

⚠️ Visibility of the new overloads

Moot — the overloads were removed.

💬 Suggestions

• SNAPSHOT_OP = "r" moved into DebeziumConstants.
• Class Javadoc on AbstractDebeziumTransformer now references DebeziumTransformerConfig.
• apply() split into named phases (selectBeforeOrAfterImage / buildOutputColumns / applyErrorTablePassthrough / applyNullabilityRules) — pure private-method extraction, no abstract methods added.
• hasSchemaField simplified via a getSourceStruct helper.
• Added the resolution-order note on the constructor.
• Kept the col.toString() LSN detection as-is, per your note.

💅 Nits

• Fixed the .s typo in HoodieTableConfig.
• The Javadoc link was already https://.
• Left the ERROR_TABLE_CURRUPT_RECORD_COL_NAME typo for a separate cleanup (it lives in BaseErrorTableWriter).

Tests

• Added: Postgres snapshot-LSN defaulting under the nested layout; MySQL _event_seq when the binlog file/pos are null (propagates null rather than fabricating a sequence).
• Reworked TestHoodieTableConfigDebeziumOrdering to cover the reconcile: flat by default, nested preserved when the caller resolved it, Postgres always _event_lsn.

Oracle / SQLServer

Oracle transformer and the chain/type-conversion transformers are planned as separate follow-up patches; SQLServer is not being upstreamed.

[rahil-c]

Thanks for the thorough review, @nsivabalan! Summary of the changes addressing the feedback (latest commit 58f39b5).

🚨 nestedDebeziumMetadataEnabled boolean / new HoodieTableConfig overloads

Removed. There's no boolean and no new overloads — inferMergingConfigsForWrites / inferMergingConfigsForV9TableCreation are back to their original signatures (unchanged vs. master) and maybeNestColumn is gone.

One deviation from the literal suggestion (resolve entirely in HoodieStreamer) that I'd like your read on: not every Debezium write path goes through the streamer. The Spark DataFrame/SQL and Flink table-creation paths reach HoodieTableConfig through HoodieTableMetaClient.init directly and rely on handlePayloadAdhocConfigs auto-correcting the ordering field to the payload's canonical column(s) — TestPayloadDeprecationFlow#testMergerBuiltinPayloadFromTableCreationPath exercises exactly that create path, and merge correctness depends on it. Resolving only in HoodieStreamer regresses those paths.

So instead of a layout flag, handlePayloadAdhocConfigs now reconciles using the existing orderingFieldName argument it already receives: it forces the flat canonical ordering (_event_lsn, or _event_bin_file,_event_pos) unless the caller already resolved the nested form — which only the Streamer does — in which case the nested form is preserved. The column knowledge lives in DebeziumConstants.resolveOrderingFields(payloadClass, nested), so the generic inference stays layout-agnostic. HoodieStreamer resolves the (flat or nested) ordering into cfg.sourceOrderingFields, so a nested table persists the nested path at creation. This also fixes a gap where the nested ordering field wasn't actually reaching table creation. Happy to iterate if you'd prefer a different split.

⚠️ Postgres nested-metadata default → false

Flipped for cross-source consistency with MySQL + backward compatibility. Removed the "defaults to true" wording from DebeziumTransformerConfig and the PostgresDebeziumTransformer Javadoc.

⚠️ Visibility of the new overloads

Moot — the overloads were removed.

💬 Suggestions

• SNAPSHOT_OP = "r" moved into DebeziumConstants.
• Class Javadoc on AbstractDebeziumTransformer now references DebeziumTransformerConfig.
• apply() split into named phases (selectBeforeOrAfterImage / buildOutputColumns / applyErrorTablePassthrough / applyNullabilityRules) — pure private-method extraction, no abstract methods added.
• hasSchemaField simplified via a getSourceStruct helper.
• Added the resolution-order note on the constructor.
• Kept the col.toString() LSN detection as-is, per your note.

💅 Nits

• Fixed the .s typo in HoodieTableConfig.
• The Javadoc link was already https://.
• Left the ERROR_TABLE_CURRUPT_RECORD_COL_NAME typo for a separate cleanup (it lives in BaseErrorTableWriter).

Tests

• Added: Postgres snapshot-LSN defaulting under the nested layout; MySQL _event_seq when the binlog file/pos are null (propagates null rather than fabricating a sequence).
• Reworked TestHoodieTableConfigDebeziumOrdering to cover the reconcile: flat by default, nested preserved when the caller resolved it, Postgres always _event_lsn.

Oracle / SQLServer

Oracle transformer and the chain/type-conversion transformers are planned as separate follow-up patches; SQLServer is not being upstreamed.

[nsivabalan]

do we have end to end functional tests for v9 for the two payloads w/ the new transformers.

If not, can we add some.

[nsivabalan]

I suggesting to move this to the call site.
why do we reconcile w/n this method?
or do we have too many call sites?

[nsivabalan]

not sure I understand this change.
I responded else where.
we should move this to call sites, were we have write configs too.
and we know if nested meta field is enabled or not.

and hence we should be able to detect the right ordering field name as well

[nsivabalan]

for your response on

🚨 nestedDebeziumMetadataEnabled boolean / new HoodieTableConfig overloads

my suggestion: these transformers are mainly meant to be used w/ HoodieStreamer only. So, our target it to get these working for HoodieStreamer and not for any other spark writers. So, we could keep it simple from adding the support.

[linliu-code]

Non-blocking: this block also runs for existing source-path users (PostgresDebeziumSource/MysqlDebeziumSource), not just the new transformer — it keys off payloadClassName. Because it overwrites cfg.sourceOrderingFields before StreamSync's pre-existing _event_seq -> file,pos normalization + ordering-field validation, a pre-v9 MySQL table with _event_seq persisted (legitimate, since handlePayloadAdhocConfigs only forces the canonical field on the v9 path) could, on restart, compare the forced _event_bin_file,_event_pos against the persisted _event_seq and hit HoodieValidationException — an upgrade break for existing users.

CI doesn't cover this: the added tests are transformer flattening + inferMergingConfigsForV9TableCreation in isolation; none constructs a HoodieStreamer on the source path or exercises the StreamSync validation, and TestPostgres/MysqlDebeziumSource only cover processDataset.

Could we add a differential check that (RecordMergeMode, ordering fields, persisted ORDERING_FIELDS, whether StreamSync validation throws) is unchanged vs master for existing source-path configs? Must-include cell: pre-v9 MySQL table, persisted _event_seq, restarted.

[linliu-code]

FYI / traceability (non-blocking): these Debezium CDC tables carry a value-level partial-update mode — handlePartialUpdateModeConfigs maps PostgresDebeziumAvroPayload → PartialUpdateMode.FILL_UNAVAILABLE (the TOAST unavailable-value back-fill). That means a schema-partial writer on the same table (e.g. a Spark MERGE INTO with a partial UPDATE SET) would emit an IS_PARTIAL log block, flip the reader to the KEEP_VALUES merger, and silently drop the configured mode for the whole file group → data corruption.

Flagging for traceability that this cross-writer risk is already guarded in the stacked #19322: CommonClientUtils.validateTableVersion now rejects shouldWritePartialUpdates() && getPartialUpdateMode().isPresent() up front (covering FILL_UNAVAILABLE / FILL_UNCHANGED / IGNORE_DEFAULTS), so it retroactively covers the Postgres path wired here too — no action needed on this PR. Noting it so the guard's coverage of the Postgres FILL_UNAVAILABLE path is discoverable from this PR.

[linliu-code]

Meanwhile, create followup tickets for these existing refactoring comments.

[linliu-code]

It means when isNested, these ordering columns will be put into the nested metadata, which may break the downstream assumption that these ordering fields are always root level fields. To tolerate this, we may have to update the table config if this pr merged like this.

[nsivabalan]

yes, this is known. this can be enabled only for new table.

[hudi-bot]

CI report:

• b7a70a5 UNKNOWN
• 7d2571b UNKNOWN
• cfdad26 UNKNOWN
• 1e51b07 UNKNOWN
• c2af7e2 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[nsivabalan]

Re-reviewed after the latest push. First, credit where it's due: 2a4926609787 + c2af7e267986 are a better resolution than what I originally asked for. Pulling the layout decision into the transformer via rootLevelOrderingColumns removed the HoodieTableConfig coupling, resolved the nested-MySQL ordering hazard @linliu-code raised (in code, rather than by documenting a "new tables only" restriction), and dropped the fragile col.toString().contains("_event_lsn") match — all at once. The hudi-common diff is now 2 new constants plus a comment typo fix, which is where it should be. The end-to-end TestDebeziumTransformerMerge coverage is a real addition too.

Overall this is net-positive and approvable. Three things below — one new finding, one I'd like to reopen, and a docs ask.

[nsivabalan]

⚠️ IMPORTANT — this ternary can stamp nullable=false onto a column that genuinely holds nulls, producing a schema that lies about the data.

Read it as a truth table, where field is a root-level column of the already-flattened dataset:

field.nullable() (root)	in nonNullableColumns?	branch	result
true	no	then	nullable=true ✅
false	no	else	nullable=false ✅
false	yes	else	nullable=false ✅
true	yes	else	nullable=false ❌

The last row is the problem. nonNullableColumns is collected from the fields inside the __data struct, but non-nullable-inside-the-struct does not imply non-null-at-root. Flattening is __data.* over

when(op == "d", col("before")).otherwise(col("after"))

and struct.* expansion over a null struct yields null for every expanded field — the inner declared nullability doesn't protect you, because the struct itself is null. Debezium emits before=null for deletes whenever Postgres REPLICA IDENTITY isn't FULL, and DEFAULT is the Postgres default. So on an ordinary configuration, every delete row has nulls in columns this method just marked non-nullable.

Nothing catches it here: createDataFrame(RDD<Row>, StructType) does not validate rows against the schema, it just attaches it. The lie surfaces later as a null-in-required-field error from the Avro/Parquet writer, pointing nowhere near this method.

Worth noting the code this replaces only ever widened — DebeziumSource.convertColumnToNullable passes non-matching fields through untouched (: field) and never narrows. The narrowing is new here.

This only bites when schema.nullable.enable=false, and the default is true, so the default path takes the early convertColumnsToNullable return and is safe. That's why this is IMPORTANT rather than blocking — but a non-default config silently emitting a wrong schema is the kind of thing that gets diagnosed months later, so I'd rather fix it than defer.

Simplest correct change is to stop narrowing — preserve whatever Spark inferred:

.map(field -> field.nullable() && !nonNullableColumns.contains(field.name())
    ? new StructField(field.name(), field.dataType(), true, field.metadata())
    : field)

That raises a design question though: the javadoc motivates this branch with _change_operation_type, but Spark already infers that non-nullable when op is non-nullable. If inference handles the cited case on its own, is the whole nonNullableColumns computation earning its keep, or could this method reduce to "widen everything, or do nothing"?

Up front: this is from code inspection — I wasn't able to confirm it at runtime (the spark3.5 profile wouldn't compile hudi-utilities against my local artifacts, unrelated to this PR). The load-bearing step is the struct.*-over-null-struct behavior; please verify that with a test rather than taking my word for it.

Suggested test — the existing two nullability tests both use a single insert with a non-null after, so neither reaches this row of the table: delete envelope with before=null, a source column declared nullable=false inside the struct, schema.nullable.enable=false. Assert the root column is still nullable, and round-trip through an actual write().save() so the writer validates the schema against the data.

[nsivabalan]

Following up to soften the framing here, and to be fair to you on provenance.

Having looked more closely at how this style of transformer is typically written, I suspect the narrowing ternary isn't something you introduced — an internal/earlier variant of this code very likely had the same shape, with the non-nullable set threaded through a shared nullability helper. If so, this is ported behavior rather than a new defect, and my "the narrowing is new here" line above is wrong. Apologies for that.

What I'd still ask, though: if the variant this came from passed an empty non-nullable set on its default path, then the narrowing branch was effectively dead in the common case, and upstreaming it as a reachable code path is a change in exposure even if the code is identical. Worth checking whether the set was ever actually populated in practice before deciding this is settled behavior.

The technical analysis is unchanged — a column non-nullable inside the before/after struct is still null at root when the selected image is null, and createDataFrame still won't catch it. Since this is an upstreaming opportunity, it seems like the right moment to fix it rather than carry it forward. Your call on whether that's this PR or a follow-up; I won't block on it.

Still code-inspection-only on my side — please confirm the struct.*-over-null-struct behavior by test before acting on it.

[nsivabalan]

⚠️ IMPORTANT — reopening this one. I don't think "pre-existing / out of scope" holds, and the consequence is worse than a degraded error message.

generateUniqueSequence validated before building the key:

if (fileId == null || fileId.trim().isEmpty() || pos == null || pos < 0) {
  throw new HoodieReadFromSourceException(
      String.format("Invalid binlog file information from Debezium: fileId=%s, pos=%s", fileId, pos));
}

All three checks are gone here, and Spark concat returns null when any argument is null — so a malformed event silently yields a null _event_seq, the ordering field. Where that lands depends on which record it is in MySqlDebeziumAvroPayload.shouldPickCurrentRecord:

• null on the incoming record → orElseThrow fires HoodieDebeziumAvroPayloadException. Still fails, but at merge time with an Avro record dumped into the message rather than fileId=null, pos=… at the source.
• null on the stored record → currentSourceSeqOpt is empty, so it hits the // handle bootstrap case branch and return false — pick the incoming record unconditionally. A null seq is indistinguishable from a legitimately-bootstrapped row, so ordering is silently skipped and an out-of-order event can overwrite a newer one.

That second path is why I'd like this addressed in the PR. It isn't just a worse message; under one arrangement it's a silent hole in the ordering guarantee that the precombine field exists to provide.

pos < 0 disappears too: a negative pos yields a well-formed-looking "000001.-5", which isCurrentSeqLatest then compares as a string and orders wrong instead of failing.

On scope — this is new code whose stated purpose is to replace the source-path flattening, so relative to the path it supersedes the guard was removed. That's a behavior regression introduced by this PR, not inherited from it. @wombatu-kun reached the same conclusion independently.

A when(col.isNull(), raise_error(...)) on the two coordinates, or an explicit validation before the concat, restores the fail-fast. Test: a MySQL envelope with source.file null (and one with source.pos null) should fail at transform time with a message naming the binlog coordinates.

[nsivabalan]

One clarification that I think explains our disagreement on "pre-existing" — and sharpens why I'd still like this addressed.

I suspect the version this was ported from also had no guard here, in which case your "pre-existing" read is entirely reasonable from that vantage point. But those are two different claims:

1. "unchanged relative to the code this was ported from" — likely true
2. "no regression in OSS" — not true, because the validation exists only in OSS, in MysqlDebeziumSource.generateUniqueSequence, and this transformer is what OSS users would adopt instead of that source

So upstreaming a variant that never had the guard into a codebase that does have it is a net loss of validation for OSS users, even though the diff faithfully reflects its origin. That's the regression I'm pointing at — not authorship.

The consequence I'd weigh most is still the second bullet in my original comment: a null _event_seq on the stored record lands in the // handle bootstrap case branch of shouldPickCurrentRecord and returns false, so ordering is silently skipped rather than failing. That's the one that seems worth not carrying forward.

Happy for this to be a follow-up if you'd rather keep the PR tight — just would like it tracked rather than closed as not-applicable.

[nsivabalan]

💬 Could you refresh the PR description before merge? It's drifted from the code across the last two commits, which makes the change hard to review from the summary alone:

• It still describes nestedDebeziumMetadataEnabled being threaded through new inferMergingConfigsForWrites / inferMergingConfigsForV9TableCreation overloads, and HoodieStreamer supplying the value — all reverted in 2a4926609787. The HoodieTableConfig diff is now just a comment typo fix.
• It says PostgresDebeziumTransformer "Nests metadata by default" and MysqlDebeziumTransformer is "Flat by default," but both now call the 3-arg super(...), so nestedFieldsEnabledByDefault=false for both. The ENABLE_NESTED_FIELDS documentation already says "both the Postgres and MySQL transformers default to flat metadata" — the description is the stale one.

One consequence worth a second look while you're in there: with both transformers now defaulting to flat, and the source.schema → db_schema_source_partition column only emitted in the nested branch, the default path no longer surfaces it — whereas PostgresDebeziumSource.processDataset always emitted it at root. It's not in META_COLUMNS and has no in-repo consumer, so blast radius is limited to external pipelines keyed on it, but the parity gap moved from an opt-in path to the default one when the nesting default flipped. Honoring hasSchemaField in the flat branch too would close it.

[hudi-agent]

⚠️ 🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

test. A few small naming nits below, but the code is generally well-structured and readable.

[hudi-agent]

🤖 nit: the class is named AbstractDebeziumTransformer but has no abstract methods — the javadoc even calls this out explicitly. Could you rename it to BaseDebeziumTransformer? The Abstract prefix signals "extend me and implement the abstract contract," which is misleading when the contract is purely constructor-based.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 nit: the field postProcessingOption encodes the type in the name — since the declared type is already Option<...>, postProcessing reads more cleanly and avoids the redundancy.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 nit: useDefaultValuesForLsnIfNull is a bit verbose and reads more like a description than a method name. Something like applySnapshotLsnDefault or defaultNullSnapshotLsn would be punchier and still self-documenting.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 Worth adding that the deferred failure is asymmetric, which strengthens the fail-fast case. In MySqlDebeziumAvroPayload.shouldPickCurrentRecord, a null _event_seq on the incoming record throws HoodieDebeziumAvroPayloadException, but a null seq on the stored record is swallowed by the bootstrap branch (!currentSourceSeqOpt.isPresent() → return false), so that record gets overwritten regardless of ordering rather than erroring. And a first-write of a malformed event goes through the insert path, so it can land silently and only trip later; preCombine would then hit Long.parseLong("null") → NumberFormatException. A guard in applySeqNo avoids all three of those less-obvious paths.