feat(cdc): Oracle Debezium CDC transformer + v9 changed-columns partial merge (stacked on #19110)#19322
Open
linliu-code wants to merge 19 commits into
Open
feat(cdc): Oracle Debezium CDC transformer + v9 changed-columns partial merge (stacked on #19110)#19322linliu-code wants to merge 19 commits into
linliu-code wants to merge 19 commits into
Conversation
Collaborator
Extracts Debezium change-event flattening out of the Debezium *source* and into a standalone, source-agnostic Transformer, so any source producing the raw Debezium envelope (Kafka, S3/file CDC logs, ...) can feed it. This first PR lands the foundation plus the Postgres implementation: - AbstractDebeziumTransformer: picks before(for deletes)/after, surfaces Debezium metadata columns, optional nested (_debezium_metadata) layout, error-table passthrough, and nullability normalization. - PostgresDebeziumTransformer: surfaces txId/lsn/xmin, defaults a null _event_lsn to 0 for snapshot rows, and nests metadata by default. - DebeziumTransformerConfig: hoodie.streamer.transformer.debezium.* configs. Output columns match DebeziumConstants, so the existing PostgresDebeziumAvroPayload keeps working unchanged. Tests: 12 unit tests (image selection, flat vs nested layout, per-subclass nested default + override, snapshot LSN defaulting).
Adds MysqlDebeziumTransformer alongside the Postgres one: surfaces the MySQL
binlog metadata (file/pos/row) as _event_bin_file/_event_pos/_event_row and
derives the _event_seq ordering column ("<binlog-suffix>.<pos>") consumed by
the existing MySqlDebeziumAvroPayload. Metadata flat by default; nested layout
supported via the shared config. Adds unit tests (flat + nested + seq).
…s enabled When the Debezium transformer nests CDC metadata under the _debezium_metadata struct (hoodie.streamer.transformer.debezium.nested.fields.enable=true), the MySQL ordering columns (_event_bin_file, _event_pos) move into that struct, so the inferred ORDERING_FIELDS must reference the nested path. handlePayloadAdhocConfigs previously hardcoded the flat names, producing a wrong ordering field in nested mode. - Add shared DebeziumConstants.DEBEZIUM_METADATA_FIELD (referenced by the transformer too). - Thread a nestedDebeziumMetadataEnabled flag via overloads of inferMergingConfigsForWrites / inferMergingConfigsForV9TableCreation (existing signatures preserved, default false; reader path and other callers unchanged). HoodieStreamer passes the real flag. - MySQL ordering fields get the _debezium_metadata. prefix when nested; Postgres _event_lsn and the operation-type delete key stay at root (transformer keeps them there), so unchanged. - Test: TestHoodieTableConfig#testInferMergingConfigsNestedDebeziumOrderingFields.
…a branches AbstractDebeziumTransformer's error-table corrupt-record passthrough and the schema.nullable.enable=false (nullability-preservation) branch had no test coverage. Add cases for both, plus a focused hudi-common test verifying HoodieTableConfig resolves the Debezium ordering field correctly whether or not nested metadata is enabled (kept in hudi-common's own test tree so its coverage is attributed to the module that owns the logic).
…overloads inferMergingConfigsForV9TableCreation/inferMergingConfigsForWrites' 5-arg overloads (kept for existing callers, delegate to the new 6-arg version with nestedDebeziumMetadataEnabled=false) had no direct test coverage.
…nt metadata-field alias - AbstractDebeziumTransformer#apply(): the final nullability rebuild only needs to force non-nullable for columns known to be non-nullable in the source row; every other column (including Debezium metadata columns) should end up nullable, regardless of their nullability in the raw envelope schema. - Remove AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD, a redundant delegate to DebeziumConstants.DEBEZIUM_METADATA_FIELD kept around only to avoid touching a few callers; reference DebeziumConstants directly instead.
… when schema.nullable.enable=false The schema-nullability rebuild marked every non-source column nullable, which flipped Debezium metadata columns (e.g. _change_operation_type) that Spark infers as non-nullable to nullable in the output schema. Restore the rule so a column stays non-nullable when Spark already infers it non-nullable or it was a non-nullable source data column, and every other column is nullable. Update TestAbstractDebeziumTransformer accordingly.
- Drop the nestedDebeziumMetadataEnabled boolean and the extra HoodieTableConfig overloads. handlePayloadAdhocConfigs now reconciles the Debezium ordering field via the existing orderingFieldName argument and DebeziumConstants.resolveOrderingFields, keeping hudi-common layout-agnostic while preserving the ordering auto-correction at the single create choke point (Spark, Flink, Streamer). - HoodieStreamer resolves the flat-or-nested Debezium ordering field into cfg.sourceOrderingFields, so a nested table persists the nested ordering path. - Default PostgresDebeziumTransformer to flat metadata, matching MySQL (cross-source consistency and backward compatibility). - Extract SNAPSHOT_OP into DebeziumConstants; split AbstractDebeziumTransformer.apply() into named phases; simplify hasSchemaField; Javadoc and comment fixes. - Cover ordering reconciliation, nested snapshot-LSN defaulting, and null MySQL binlog coordinates in tests.
…s partial merge Adds Oracle Debezium CDC support to OSS, stacked on the Postgres/MySQL transformer foundation (apache#19110). OSS counterpart of the internal change. hudi-common: - OracleDebeziumAvroPayload: Oracle CDC payload (legacy merge + the identifier the v9 table-config inference keys on). - DebeziumConstants: Oracle source/flattened columns + _event_ordering / _changed_columns; resolveOrderingFields returns _event_ordering (root) for the Oracle payload. - PartialUpdateMode.FILL_UNCHANGED + PartialUpdateHandler.reconcileChangedColumns: changed-columns-driven partial update (preserves prior values for unchanged columns of ANY type under PK-only supplemental logging), unioning the changed-columns set across records, with toasted-sentinel + retain-metadata handling. Implemented against HoodieSchema. - HoodieTableConfig: infer EVENT_TIME_ORDERING + FILL_UNCHANGED + _event_ordering + merge properties for OracleDebeziumAvroPayload on table version 9. hudi-utilities: - OracleDebeziumTransformer: flattens the Oracle envelope, surfaces SCN metadata + composite _event_ordering, computes _changed_columns (toasted-excluded), sets _hoodie_is_deleted, filters unsupported ops. Tests: TestOracleDebeziumTransformer, TestOracleDebeziumAvroPayload (38), TestPartialUpdateHandler FILL_UNCHANGED unit cases (11), TestHoodieTableConfigDebeziumOrdering Oracle inference cases, and a v9 read-path IT (TestOracleDebeziumV9ReadMerge). hudi-common verified locally: 61 tests pass, checkstyle 0, rat 0, scalastyle 0.
…CHANGED applies end-to-end Root cause (read path): FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging never added the partial-update changed-columns field to the required read schema, so column pruning dropped _changed_columns. reconcileChangedColumns then read a null list and fell back to newer-wins, so every unchanged column took the incoming placeholder. IGNORE_DEFAULTS / FILL_UNAVAILABLE are unaffected because they inspect the data columns' own values rather than a separate list column. Fix 1 (schema): retain the configured changed-columns field as a mandatory merge field so pruning keeps it. Fix 2 (merge value type): the synthesized union changed-columns value used convertValueToEngineType, which on Spark returns the comparison wrapper and blows up unsafe-row projection / inline compaction with a ClassCastException. Reuse a record's raw engine-native value when it already equals the union; only for genuinely-disjoint updates synthesize via convertPartitionValueToEngineType (unwrapped engine storage). Adds IT scenarios (delete, inline compaction, nested metadata struct) and a two-step write-path inference regression test. Verified: full v9 IT green (6/6); TestPartialUpdateHandler (17) + TestHoodieTableConfigDebeziumOrdering (7) green; checkstyle/rat/scalastyle clean.
…gineType contract Rebased the stack onto current apache/hudi master, which since the branch's base: - moved HoodieAvroUtils and AvroRecordContext from org.apache.hudi.avro to org.apache.hudi.common.avro (update the two imports), and - redefined convertValueToEngineType to return the engine-native STORAGE form (e.g. an unwrapped Spark UTF8String), splitting the comparison-wrapping into ensureComparability/convertOrderingValueToEngineType. So the synthesized union changed-columns value now uses convertValueToEngineType (was the interim convertPartitionValueToEngineType, which no longer exists on master). Verified against rebased master: hudi-common Oracle unit tests (62) green; TestOracleDebeziumV9ReadMerge IT 6/6 green; checkstyle/rat/scalastyle clean.
linliu-code
force-pushed
the
eng-44204-oss-oracle
branch
2 times, most recently
from
July 19, 2026 01:27
9dec96e to
feb4832
Compare
linliu-code
changed the title
linliu-code
marked this pull request as ready for review
July 19, 2026 02:27
…og-file + guard coverage Hardens the v9 FILL_UNCHANGED partial merge: - FILL_UNCHANGED <-> IS_PARTIAL was an unguarded, silently-corrupting assumption. A schema-partial log block (written when hoodie.write.partial.update.schema is set, e.g. a Spark MERGE INTO with a partial UPDATE SET) flips the reader to the KEEP_VALUES merger and drops the changed-columns logic for the whole file group, letting CDC placeholder values win. HoodieAppendHandle now rejects a partial-update write against a FILL_UNCHANGED table up front (fail fast instead of corrupt on read). Scoped to FILL_UNCHANGED so other modes are unaffected. - COW was untested and is a distinct write-time merge path. Added a COW variant; it passes (COW applies the partial merge at write time, no log files) with the unchanged column preserved. - MOR tests could pass vacuously without exercising merge-on-read. Added logFileCount() assertions: the update test requires >=1 log file, the disjoint-union test requires >=2. - Documented the FILL_UNCHANGED contract on the enum (full-schema value-level records; all-columns- exist rests on the transformer's full-schema guarantee; incompatibility with partial-update writes). - Added a guard verification test: a partial-update write on a FILL_UNCHANGED table throws. Verified against current master: TestOracleDebeziumV9ReadMerge 8/8 green; checkstyle + scalastyle clean.
Closes the last coverage gap (FILL_UNCHANGED x schema evolution was untested). reconcileChangedColumns iterates the incoming schema and reads unchanged columns from the older base schema, so a base != incoming schema merge was unexercised. New IT: base has no `email` column; an evolved update (schema-on-read enabled) adds it and lists only `email` in _changed_columns. Asserts the pre-existing unchanged columns are preserved from the base and the new column takes the incoming value. Passes — confirms correctness. TestOracleDebeziumV9ReadMerge 9/9 green; scalastyle clean.
…pdate modes The FILL_UNCHANGED <-> IS_PARTIAL silent-corruption conflict is not FILL_UNCHANGED-specific: BufferedRecordMergerFactory drops the partialUpdateMode generically (Option.empty()) whenever a schema-partial IS_PARTIAL log block is present, so FILL_UNAVAILABLE and IGNORE_DEFAULTS are dropped to KEEP_VALUES the same way, letting the sentinel/default win over the prior value. This was a pre-existing gap for those modes; only FILL_UNCHANGED was guarded. Move + generalize the guard into CommonClientUtils.validateTableVersion (the canonical write-time validation hook, already checking shouldWritePartialUpdates, invoked for COW and MOR via BaseHoodieWriteClient): reject a partial-update write against a table with ANY value-level partial-update mode. Removes the narrower MOR-only guard from HoodieAppendHandle. No intended combination is broken (partial-update writes originate only from Spark MERGE INTO). Adds 3 unit tests (reject with FILL_UNAVAILABLE to prove generalization; allow full-schema write on a value-mode table; allow partial write when no mode); the end-to-end guard IT still passes via the relocated guard. Verified against master: TestCommonClientUtils 88/88, TestOracleDebeziumV9ReadMerge 9/9; checkstyle clean.
… merge paths Generated-sequence differential test: random Oracle-CDC event sequences (insert + 2-4 updates changing random, often disjoint, column subsets; unchanged columns carry a placeholder that must not win) run through three merge paths -- MOR snapshot (log-vs-log deltaMerge), MOR inline compaction, COW write-time merge -- each diffed against a reference model applying FILL_UNCHANGED semantics directly (changed take incoming, unchanged keep prior, in _event_ordering order). Any path disagreeing with the reference or with each other is a merge bug; catches the union bug and COW/MOR path divergence by construction. Seeded (8 sequences). All 3 paths match the reference. TestOracleDebeziumV9ReadMerge differential test green; scalastyle clean.
…h mid-commit) INV7 idempotency: re-deliver the latest event exactly, then stale re-deliver an older (ord=200) event after a newer (ord=300) one; assert the merged read is unchanged. EVENT_TIME_ORDERING + FILL_UNCHANGED is idempotent under duplicate/out-of-order delivery. INV8 crash mid-commit: revert the last completed delta-commit to inflight via the timeline API (revertToInflight -- data on disk, completion marker gone), then assert the reader ignores it (the crashed update is invisible; read reflects only committed state). Faithful single-JVM stand-in for the data-written-but-uncommitted window; a real process kill+restart would be a separate harness. TestOracleDebeziumV9ReadMerge fault-injection tests green; scalastyle clean.
…L_UNCHANGED (INV8)
The existing INV8 test uses revertToInflight inside one JVM, standing in for the
"data written, completion marker missing" window only. This adds a real crash: a
Spark-writing subprocess launched via `java -cp`, SIGKILL'd (Process.destroyForcibly)
as it races the update-commit, then a retry that must drain and recover.
Per fresh table dir (independent samples), it asserts the crashed table is readable
and atomic -- exactly one row, name in {alice, bob}, amount == the committed value,
never torn/half-merged/placeholder-leaked -- and that a retry drains the pending
update to the correct final state with the unchanged column preserved.
Gated off by default (assumeTrue on -Dhudi.crash.harness=true): it launches a real
subprocess (~40s) and depends on classpath replay + kill timing, both fragile in a
CI matrix, so it belongs in a soak/nightly lane rather than the PR gate. Verified
locally: passes with the flag (~44s), skips without it (~1s).
…ouble-fault nemesis Builds on the P1 real-SIGKILL harness with deterministic-window kills and two new failure modes, all non-invasive (no write-path changes): - Precise kill (supersedes P1's coarse marker): SIGKILL the writer the instant the update's .deltacommit.inflight lands on the timeline -- the AFTER_INFLIGHT_BEFORE_COMPLETION window. Asserts atomicity + recovery per fresh dir. - Crash during inline compaction: the FILL_UNCHANGED base a half-done compaction would produce must never be visible; recovery re-runs and converges. Asserts the nemesis armed (a .compaction instant was seen). - Double fault (unreachable in one JVM): crash 1 leaves a pending update inflight; the recovery writer rolls it back; crash 2 kills it around that rollback; a third writer must still converge. Asserts a .rollback instant was observed. The double-fault uses a large (100k-row) update so the post-inflight data-write lasts long enough for the kill to reliably land before completion; a 1-row commit completes too fast to catch without a production phase hook. A deterministic AFTER_INFLIGHT_BEFORE_COMPLETION freeze still needs that hook, along with Kafka at-least-once replay and real-cluster executor-mid-write kills (remaining work). Opt-in soak only (-Dhudi.crash.harness=true). Verified: 4/4 pass with the flag (~147s), 4/4 skip without it; scalastyle + checkstyle clean.
…L_UNCHANGED
Runs the real end-to-end streaming crash test the spark-side harness cannot cover: a
HoodieStreamer subprocess ingests Oracle-Debezium envelopes from a Kafka topic through the
OracleDebeziumTransformer into a v9 FILL_UNCHANGED MOR table, with source-checkpointed
at-least-once delivery. SIGKILL'ing and restarting the streamer exercises:
- INV8 (crash mid-commit): after a real kill the table is readable and shows only a
committed prefix -- never torn, and notes never leaks the toasted placeholder;
- INV7 (at-least-once replay): the restart resumes from the committed Kafka checkpoint,
replays the uncommitted tail, and converges to the terminal state with exactly one row
per key (no duplication, no loss) and FILL_UNCHANGED preserved across the crash.
A probe first validates the whole pipeline (two commits + checkpoint resume) before the
kill variant.
Design choices (deliberate):
- JSON on the wire (JsonKafkaSource) rather than schema-registry Avro: a subprocess cannot
reach an in-JVM mock schema registry, and the wire format is orthogonal to the
replay/idempotency invariants -- transformer, payload, v9 merge, checkpoint and
crash/restart are all the real production paths.
- A source-only schema provider (null target schema) so StreamSync deduces the writer
schema from the transformer's flattened output rather than the nested envelope schema.
- MOR compaction: once-mode requires hoodie.compact.inline=true, continuous-mode requires
it false (async compaction); set by mode.
Opt-in soak only (-Dhudi.crash.harness=true); needs a Docker daemon (testcontainers Kafka).
Verified: 2/2 pass with the flag (~61s), 2/2 skip without it; checkstyle clean.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
Adds an Oracle Debezium CDC transformer and table-version-9 partial-merge support to Hudi, so Oracle CDC tables can merge on the built-in
EVENT_TIME_ORDERING+ partial-update path (driven by the_changed_columnslist) instead of relying on a custom payload class. Stacked on #19110 (Postgres/MySQL Debezium transformers); the Oracle commits are the top of the stack.Summary and Changelog
hudi-utilities: addOracleDebeziumTransformer(+ sharedAbstractDebeziumTransformer) alongside the Postgres/MySQL transformers from feat(utilities): add Postgres and Mysql Debezium CDC transformers #19110; wire theDebeziumTransformerConfignested-fields option.hudi-common: addOracleDebeziumAvroPayload, Oracle constants inDebeziumConstants, and a newPartialUpdateMode.FILL_UNCHANGEDstrategy inPartialUpdateHandlerthat preserves unchanged columns using the CDC_changed_columnslist (unions both records' changed sets across log-vs-log merges).hudi-common:HoodieTableConfigv9 inference maps the Oracle payload toEVENT_TIME_ORDERING+FILL_UNCHANGED+ the changed/retain/delete merge properties.hudi-common:FileGroupReaderSchemaHandlerretains the configured changed-columns field as a mandatory merge column so pruning keeps it (without this, unchanged columns fall back to the incoming placeholder on read).FILL_UNCHANGEDhandler; an end-to-endTestOracleDebeziumV9ReadMergespark IT (update / null / disjoint-union / delete / inline-compaction / nested-metadata).No code was copied from external sources; the transformer/payload mirror the existing onehouse-dataplane implementations.
Impact
New opt-in CDC transformer + a new partial-update mode. No change to existing merge modes or payloads.
FILL_UNCHANGEDonly engages for tables configured with the Oracle payload / changed-columns merge property.Risk Level
low — additive; gated behind the Oracle payload/config. Verified against current master:
hudi-commonOracle unit tests (62) green,TestOracleDebeziumV9ReadMergeIT 6/6 green, checkstyle/rat/scalastyle clean. (hudi-utilitiestransformer compile relies on CI due to a local-only Confluent dependency resolution issue.)Documentation Update
none (internal CDC wiring; no new user-facing website config in this PR).
Contributor's checklist