feat(cdc): Oracle Debezium CDC transformer + v9 changed-columns partial merge (stacked on #19110)

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/CommonClientUtils.java

@@ -57,6 +57,19 @@ public static void validateTableVersion(HoodieTableConfig tableConfig, HoodieWri
       throw new HoodieNotSupportedException("Partial updates are not supported for table versions < 8. "
           + "Please unset " + HoodieWriteConfig.WRITE_PARTIAL_UPDATE_SCHEMA.key());
     }
+    // A table configured with a value-level partial-update merge mode (IGNORE_DEFAULTS / FILL_UNAVAILABLE /
+    // FILL_UNCHANGED) expects FULL-SCHEMA records whose unchanged columns carry a default/sentinel/placeholder
+    // that the read-side merge resolves. A schema-partial write (WRITE_PARTIAL_UPDATE_SCHEMA, e.g. a Spark
+    // MERGE INTO with a partial UPDATE SET) writes an IS_PARTIAL log block, which flips the reader to the
+    // KEEP_VALUES partial merger and silently drops the configured mode for the whole file group -- letting the
+    // placeholder/sentinel/default win over the prior value (silent corruption). Reject the combination up front.
+    if (writeConfig.shouldWritePartialUpdates() && tableConfig.getPartialUpdateMode().isPresent()) {
+      throw new HoodieNotSupportedException(String.format(
+          "Partial-update writes (%s) are incompatible with a table configured for value-level partial-update "
+              + "merge (mode=%s); such tables require full-schema writes. Unset %s or use a full-schema writer.",
+          HoodieWriteConfig.WRITE_PARTIAL_UPDATE_SCHEMA.key(), tableConfig.getPartialUpdateMode().get(),
+          HoodieWriteConfig.WRITE_PARTIAL_UPDATE_SCHEMA.key()));
+    }
 
     if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT) && writeConfig.isNonBlockingConcurrencyControl()) {
       throw new HoodieNotSupportedException("Non-blocking concurrency control is not supported for table versions < 8.");

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/utils/TestCommonClientUtils.java

@@ -22,6 +22,8 @@
 
 import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.PartialUpdateMode;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieNotSupportedException;
 import org.apache.hudi.util.CommonClientUtils;
@@ -54,6 +56,50 @@ void testDisallowPartialUpdatesPreVersion8() {
     assertThrows(HoodieNotSupportedException.class, () -> CommonClientUtils.validateTableVersion(tConfig, wConfig));
   }
 
+  @Test
+  void testDisallowPartialUpdateWriteOnValueLevelPartialUpdateMode() {
+    // A schema-partial write (WRITE_PARTIAL_UPDATE_SCHEMA) on a table configured with ANY value-level
+    // partial-update mode must be rejected (it would flip the reader to KEEP_VALUES and silently drop the
+    // mode). Uses FILL_UNAVAILABLE to prove the guard covers all such modes, not just FILL_UNCHANGED.
+    HoodieWriteConfig wConfig = mock(HoodieWriteConfig.class);
+    HoodieTableConfig tConfig = mock(HoodieTableConfig.class);
+    when(wConfig.getWriteVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(tConfig.getTableVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(wConfig.shouldWritePartialUpdates()).thenReturn(true);
+    when(tConfig.getPartialUpdateMode()).thenReturn(Option.of(PartialUpdateMode.FILL_UNAVAILABLE));
+
+    assertThrows(HoodieNotSupportedException.class, () -> CommonClientUtils.validateTableVersion(tConfig, wConfig));
+  }
+
+  @Test
+  void testAllowPartialUpdateWriteWhenNoPartialUpdateMode() {
+    // Partial-update writes are fine when the table has no value-level partial-update mode configured.
+    HoodieWriteConfig wConfig = mock(HoodieWriteConfig.class);
+    HoodieTableConfig tConfig = mock(HoodieTableConfig.class);
+    when(wConfig.getWriteVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(tConfig.getTableVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(wConfig.shouldWritePartialUpdates()).thenReturn(true);
+    when(tConfig.getPartialUpdateMode()).thenReturn(Option.empty());
+
+    // Should not throw.
+    CommonClientUtils.validateTableVersion(tConfig, wConfig);
+  }
+
+  @Test
+  void testAllowFullSchemaWriteOnValueLevelPartialUpdateMode() {
+    // A full-schema write (no WRITE_PARTIAL_UPDATE_SCHEMA) on a value-level partial-update table is the
+    // normal CDC ingestion path and must be allowed.
+    HoodieWriteConfig wConfig = mock(HoodieWriteConfig.class);
+    HoodieTableConfig tConfig = mock(HoodieTableConfig.class);
+    when(wConfig.getWriteVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(tConfig.getTableVersion()).thenReturn(HoodieTableVersion.NINE);
+    when(wConfig.shouldWritePartialUpdates()).thenReturn(false);
+    when(tConfig.getPartialUpdateMode()).thenReturn(Option.of(PartialUpdateMode.FILL_UNCHANGED));
+
+    // Should not throw.
+    CommonClientUtils.validateTableVersion(tConfig, wConfig);
+  }
+
   @Test
   void testDisallowNBCCPreVersion8() {
     // given:

hudi-common/src/main/java/org/apache/hudi/common/model/debezium/DebeziumConstants.java

@@ -48,6 +48,10 @@ public class DebeziumConstants {
   public static final String INCOMING_SOURCE_LSN_FIELD = "source.lsn";
   public static final String INCOMING_SOURCE_XMIN_FIELD = "source.xmin";
 
+  // INPUT COLUMNS SPECIFIC TO ORACLE
+  public static final String INCOMING_SOURCE_SCN_FIELD = "source.scn";
+  public static final String INCOMING_SOURCE_COMMIT_SCN_FIELD = "source.commit_scn";
+
   // OUTPUT COLUMNS
   public static final String FLATTENED_OP_COL_NAME = "_change_operation_type";
   public static final String UPSTREAM_PROCESSING_TS_COL_NAME = "_upstream_event_processed_ts_ms";
@@ -66,8 +70,23 @@ public class DebeziumConstants {
   public static final String FLATTENED_LSN_COL_NAME = "_event_lsn";
   public static final String FLATTENED_XMIN_COL_NAME = "_event_xmin";
 
+  // OUTPUT COLUMNS SPECIFIC TO ORACLE
+  public static final String FLATTENED_SCN_COL_NAME = "_event_scn";
+  public static final String FLATTENED_COMMIT_SCN_COL_NAME = "_event_commit_scn";
+  // Composite event-time ordering column for Oracle: zero-padded "commit_scn.scn", materialized at
+  // the root level by the transformer (so it is not nested even when metadata nesting is enabled).
+  public static final String FLATTENED_ORDERING_COL_NAME = "_event_ordering";
+  // Comma-separated list of data columns that changed in an Oracle CDC update event.
+  public static final String CHANGED_COLUMNS_FIELD = "_changed_columns";
+
   // Other Constants
   public static final String DELETE_OP = "d";
+  // Snapshot ("read") operation emitted by Debezium for rows captured during an initial or
+  // incremental snapshot rather than from the transaction log.
+  public static final String SNAPSHOT_OP = "r";
+
+  // Struct column under which the Debezium metadata columns are grouped when nested fields are enabled.
+  public static final String DEBEZIUM_METADATA_FIELD = "_debezium_metadata";
 
   // List of meta data columns
   public static List<String> META_COLUMNS = Collections.unmodifiableList(Arrays.asList(
@@ -79,5 +98,37 @@ public class DebeziumConstants {
       FLATTENED_XMIN_COL_NAME,
       FLATTENED_SHARD_NAME
   ));
+
+  /**
+   * Resolves the canonical ordering field(s) for a Debezium payload, or {@code null} for any other
+   * payload (in which case the caller keeps its own ordering field). This is the single source of
+   * truth for the ordering field that the {@code *DebeziumAvroPayload} merge relies on, shared by
+   * every write path that creates a Debezium table (Hudi Streamer, Spark, Flink).
+   *
+   * <p>The Postgres LSN and the operation-type column stay at the root level even when the
+   * transformer groups the CDC metadata under the {@link #DEBEZIUM_METADATA_FIELD} struct, so the
+   * Postgres ordering field is always {@code _event_lsn}. The MySQL binlog coordinates move into that
+   * struct when nesting is enabled, so the MySQL ordering field is resolved against the nested path.
+   *
+   * @param payloadClassName fully-qualified payload class name (may be {@code null}).
+   * @param nestedMetadataEnabled whether the Debezium transformer nests the CDC metadata columns
+   *                              under the {@link #DEBEZIUM_METADATA_FIELD} struct.
+   * @return the ordering field(s) for a Debezium payload, or {@code null} for any other payload.
+   */
+  public static String resolveOrderingFields(String payloadClassName, boolean nestedMetadataEnabled) {
+    if (PostgresDebeziumAvroPayload.class.getName().equals(payloadClassName)) {
+      return FLATTENED_LSN_COL_NAME;
+    }
+    if (MySqlDebeziumAvroPayload.class.getName().equals(payloadClassName)) {
+      String prefix = nestedMetadataEnabled ? DEBEZIUM_METADATA_FIELD + "." : "";
+      return prefix + FLATTENED_FILE_COL_NAME + "," + prefix + FLATTENED_POS_COL_NAME;
+    }
+    if (OracleDebeziumAvroPayload.class.getName().equals(payloadClassName)) {
+      // The Oracle composite ordering column is materialized at the root level regardless of metadata
+      // nesting, so it needs no _debezium_metadata prefix.
+      return FLATTENED_ORDERING_COL_NAME;
+    }
+    return null;
+  }
 }