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

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

@@ -69,6 +69,9 @@ public class DebeziumConstants {
   // Other Constants
   public static final String DELETE_OP = "d";
 
+  // 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(
       FLATTENED_OP_COL_NAME,

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java

@@ -849,6 +849,16 @@ public static Map<String, String> inferMergingConfigsForV9TableCreation(RecordMe
                                                                           String recordMergeStrategyId,
                                                                           String orderingFieldName,
                                                                           HoodieTableVersion tableVersion) {
+    return inferMergingConfigsForV9TableCreation(recordMergeMode, payloadClassName, recordMergeStrategyId,
+        orderingFieldName, tableVersion, false);
+  }
+
+  public static Map<String, String> inferMergingConfigsForV9TableCreation(RecordMergeMode recordMergeMode,
+                                                                          String payloadClassName,
+                                                                          String recordMergeStrategyId,
+                                                                          String orderingFieldName,
+                                                                          HoodieTableVersion tableVersion,
+                                                                          boolean nestedDebeziumMetadataEnabled) {
     Map<String, String> reconciledConfigs = new HashMap<>();
     if (tableVersion.lesserThan(HoodieTableVersion.NINE)) {
       throw new HoodieIOException("Unsupported flow for table versions less than 9");
@@ -894,7 +904,7 @@ public static Map<String, String> inferMergingConfigsForV9TableCreation(RecordMe
         // Additional custom merge properties.s
         // Certain payloads are migrated to non payload way from 1.1 Hudi binary and the reader might need certain properties for the
         // merge to function as expected. Handing such special cases here.
-        handlePayloadAdhocConfigs(payloadClassName, reconciledConfigs);
+        handlePayloadAdhocConfigs(payloadClassName, reconciledConfigs, nestedDebeziumMetadataEnabled);
       }
     }
     return reconciledConfigs;
@@ -922,24 +932,39 @@ private static void handlePartialUpdateModeConfigs(String payloadClassName, Map<
     }
   }
 
-  private static void handlePayloadAdhocConfigs(String payloadClassName, Map<String, String> reconciledConfigs) {
+  private static void handlePayloadAdhocConfigs(String payloadClassName, Map<String, String> reconciledConfigs,
+                                                boolean nestedDebeziumMetadataEnabled) {
     // Certain payloads are migrated to non payload way from 1.1 Hudi binary and the reader might need certain properties for the
     // merge to function as expected. Handing such special cases here.
     if (payloadClassName.equals(PostgresDebeziumAvroPayload.class.getName())) {
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + PARTIAL_UPDATE_UNAVAILABLE_VALUE, DEBEZIUM_UNAVAILABLE_VALUE);
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY, DebeziumConstants.FLATTENED_OP_COL_NAME);
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER, DebeziumConstants.DELETE_OP);
+      // The Postgres ordering field (_event_lsn) and the operation-type column are kept at the root level
+      // even when nested fields are enabled, so they need no _debezium_metadata prefix.
       reconciledConfigs.put(ORDERING_FIELDS.key(), DebeziumConstants.FLATTENED_LSN_COL_NAME);
     } else if (payloadClassName.equals(MySqlDebeziumAvroPayload.class.getName())) {
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY, DebeziumConstants.FLATTENED_OP_COL_NAME);
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER, DebeziumConstants.DELETE_OP);
-      reconciledConfigs.put(ORDERING_FIELDS.key(), DebeziumConstants.FLATTENED_FILE_COL_NAME + "," + DebeziumConstants.FLATTENED_POS_COL_NAME);
+      // The MySQL ordering fields (_event_bin_file, _event_pos) are moved into the _debezium_metadata struct
+      // when nested fields are enabled, so the ordering field config must reference the nested path.
+      reconciledConfigs.put(ORDERING_FIELDS.key(),
+          maybeNestColumn(DebeziumConstants.FLATTENED_FILE_COL_NAME, nestedDebeziumMetadataEnabled)
+              + "," + maybeNestColumn(DebeziumConstants.FLATTENED_POS_COL_NAME, nestedDebeziumMetadataEnabled));
     } else if (payloadClassName.equals(AWSDmsAvroPayload.class.getName())) {
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY, OP_FIELD);
       reconciledConfigs.put(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER, DELETE_OPERATION_VALUE);
     }
   }
 
+  /**
+   * Prefixes a flattened Debezium column with the {@link DebeziumConstants#DEBEZIUM_METADATA_FIELD} struct
+   * path when nested fields are enabled; returns the column unchanged otherwise.
+   */
+  private static String maybeNestColumn(String column, boolean nestedDebeziumMetadataEnabled) {
+    return nestedDebeziumMetadataEnabled ? DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + column : column;
+  }
+
   /**
    * To be invoked for table creation flows or writer flows.
    * @return the merging configs to use.
@@ -949,9 +974,24 @@ public static Triple<RecordMergeMode, String, String> inferMergingConfigsForWrit
                                                                                      String recordMergeStrategyId,
                                                                                      String orderingFieldNamesAsString,
                                                                                      HoodieTableVersion tableVersion) {
+    return inferMergingConfigsForWrites(recordMergeMode, payloadClassName, recordMergeStrategyId,
+        orderingFieldNamesAsString, tableVersion, false);
+  }
+
+  /**
+   * @param nestedDebeziumMetadataEnabled when true, Debezium CDC metadata columns are nested under the
+   *     {@link DebeziumConstants#DEBEZIUM_METADATA_FIELD} struct, so the inferred ordering field(s) for
+   *     Debezium payloads must be resolved against that nested path.
+   */
+  public static Triple<RecordMergeMode, String, String> inferMergingConfigsForWrites(RecordMergeMode recordMergeMode,
+                                                                                     String payloadClassName,
+                                                                                     String recordMergeStrategyId,
+                                                                                     String orderingFieldNamesAsString,
+                                                                                     HoodieTableVersion tableVersion,
+                                                                                     boolean nestedDebeziumMetadataEnabled) {
     if (tableVersion.greaterThanOrEquals(HoodieTableVersion.NINE)) {
       Map<String, String> inferredMergingConfigs = inferMergingConfigsForV9TableCreation(
-          recordMergeMode, payloadClassName, recordMergeStrategyId, orderingFieldNamesAsString, tableVersion);
+          recordMergeMode, payloadClassName, recordMergeStrategyId, orderingFieldNamesAsString, tableVersion, nestedDebeziumMetadataEnabled);
       checkArgument(inferredMergingConfigs.containsKey(HoodieTableConfig.RECORD_MERGE_MODE.key()));
       return Triple.of(
           RecordMergeMode.valueOf(inferredMergingConfigs.get(RECORD_MERGE_MODE.key())),

hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.table;
+
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.debezium.DebeziumConstants;
+import org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload;
+import org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload;
+import org.apache.hudi.common.util.collection.Triple;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests that {@link HoodieTableConfig#inferMergingConfigsForV9TableCreation} resolves the
+ * Debezium ordering field against the {@code _debezium_metadata} struct path when nested metadata
+ * is enabled. Kept in {@code hudi-common}'s own test source set (rather than alongside the larger
+ * {@code TestHoodieTableConfig} suite in {@code hudi-hadoop-common}) so coverage for this
+ * hudi-common-owned logic is attributed to the module that owns it.
+ */
+class TestHoodieTableConfigDebeziumOrdering {
+
+  @Test
+  void mysqlOrderingFieldsAreNestedWhenDebeziumMetadataIsNested() {
+    Map<String, String> flat = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, false);
+    assertEquals(DebeziumConstants.FLATTENED_FILE_COL_NAME + "," + DebeziumConstants.FLATTENED_POS_COL_NAME,
+        flat.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+
+    Map<String, String> nested = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, true);
+    assertEquals(
+        DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_FILE_COL_NAME + ","
+            + DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_POS_COL_NAME,
+        nested.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+  }
+
+  @Test
+  void postgresOrderingFieldStaysAtRootRegardlessOfNesting() {
+    Map<String, String> flat = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, PostgresDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, false);
+    Map<String, String> nested = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, PostgresDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, true);
+
+    assertEquals(DebeziumConstants.FLATTENED_LSN_COL_NAME, flat.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+    assertEquals(DebeziumConstants.FLATTENED_LSN_COL_NAME, nested.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+  }
+
+  @Test
+  void fiveArgOverloadsDelegateToNestedFalse() {
+    // The pre-existing 5-arg inferMergingConfigsForV9TableCreation/inferMergingConfigsForWrites
+    // overloads must keep behaving exactly as if nestedDebeziumMetadataEnabled=false was passed,
+    // since existing callers (e.g. the reader path) are unaware of Debezium metadata nesting.
+    Map<String, String> viaFiveArg = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE);
+    Map<String, String> viaSixArgFalse = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, false);
+    assertEquals(viaSixArgFalse.get(HoodieTableConfig.ORDERING_FIELDS.key()), viaFiveArg.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+
+    Triple<RecordMergeMode, String, String> writesViaFiveArg = HoodieTableConfig.inferMergingConfigsForWrites(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE);
+    Triple<RecordMergeMode, String, String> writesViaSixArgFalse = HoodieTableConfig.inferMergingConfigsForWrites(
+        null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, false);
+    // Compare components individually: the payload class (middle) is legitimately null for V9
+    // tables, and Triple#equals NPEs on a null middle rather than short-circuiting.
+    assertEquals(writesViaSixArgFalse.getLeft(), writesViaFiveArg.getLeft());
+    assertEquals(writesViaSixArgFalse.getMiddle(), writesViaFiveArg.getMiddle());
+    assertEquals(writesViaSixArgFalse.getRight(), writesViaFiveArg.getRight());
+  }
+}

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java

@@ -28,6 +28,7 @@
 import org.apache.hudi.common.model.OverwriteNonDefaultsWithLatestAvroPayload;
 import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
 import org.apache.hudi.common.model.PartialUpdateAvroPayload;
+import org.apache.hudi.common.model.debezium.DebeziumConstants;
 import org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload;
 import org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload;
 import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
@@ -734,6 +735,36 @@ void testInferMergingConfigsForV9TableCreation(String testName, RecordMergeMode
     }
   }
 
+  @Test
+  void testInferMergingConfigsNestedDebeziumOrderingFields() {
+    String mysqlPayload = MySqlDebeziumAvroPayload.class.getName();
+    String pgPayload = PostgresDebeziumAvroPayload.class.getName();
+
+    // MySQL, flat (default): ordering fields are the root-level binlog columns.
+    Map<String, String> mysqlFlat = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, mysqlPayload, null, "ts", HoodieTableVersion.NINE, false);
+    assertEquals(DebeziumConstants.FLATTENED_FILE_COL_NAME + "," + DebeziumConstants.FLATTENED_POS_COL_NAME,
+        mysqlFlat.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+
+    // MySQL, nested: the binlog columns live under the _debezium_metadata struct, so the ordering
+    // fields must be resolved against that nested path.
+    Map<String, String> mysqlNested = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, mysqlPayload, null, "ts", HoodieTableVersion.NINE, true);
+    assertEquals(
+        DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_FILE_COL_NAME + ","
+            + DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_POS_COL_NAME,
+        mysqlNested.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+
+    // Postgres: the ordering field (_event_lsn) is kept at the root level even when nested fields are
+    // enabled, so it is unchanged in both cases.
+    Map<String, String> pgFlat = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, pgPayload, null, "ts", HoodieTableVersion.NINE, false);
+    Map<String, String> pgNested = HoodieTableConfig.inferMergingConfigsForV9TableCreation(
+        null, pgPayload, null, "ts", HoodieTableVersion.NINE, true);
+    assertEquals(DebeziumConstants.FLATTENED_LSN_COL_NAME, pgFlat.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+    assertEquals(DebeziumConstants.FLATTENED_LSN_COL_NAME, pgNested.get(HoodieTableConfig.ORDERING_FIELDS.key()));
+  }
+
   @Test
   void testInferMergingConfigsForVersion9WithInconsistentConfigs() {
     // Test case: Inconsistent merge mode and strategy should throw exception

hudi-utilities/src/main/java/org/apache/hudi/utilities/config/DebeziumTransformerConfig.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.utilities.config;
+
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.hudi.common.config.HoodieConfig;
+
+import javax.annotation.concurrent.Immutable;
+
+import static org.apache.hudi.common.util.ConfigUtils.DELTA_STREAMER_CONFIG_PREFIX;
+import static org.apache.hudi.common.util.ConfigUtils.STREAMER_CONFIG_PREFIX;
+
+/**
+ * Configurations controlling the Debezium CDC transformers (e.g.
+ * {@code PostgresDebeziumTransformer}, {@code MysqlDebeziumTransformer}).
+ */
+@Immutable
+@ConfigClassProperty(name = "Debezium Transformer Configs",
+    groupName = ConfigGroups.Names.HUDI_STREAMER,
+    subGroupName = ConfigGroups.SubGroupNames.NONE,
+    description = "Configurations controlling the Debezium CDC transformers that flatten "
+        + "Debezium change-event envelopes into Hudi rows.")
+public class DebeziumTransformerConfig extends HoodieConfig {
+
+  private static final String PREFIX = STREAMER_CONFIG_PREFIX + "transformer.debezium.";
+  private static final String OLD_PREFIX = DELTA_STREAMER_CONFIG_PREFIX + "transformer.debezium.";
+
+  public static final ConfigProperty<Boolean> ENABLE_NESTED_FIELDS = ConfigProperty
+      .key(PREFIX + "nested.fields.enable")
+      .defaultValue(false)
+      .withAlternatives(OLD_PREFIX + "nested.fields.enable")
+      .markAdvanced()
+      .sinceVersion("1.3.0")
+      .withDocumentation("When enabled, the Debezium transformer packs the CDC metadata columns "
+          + "under a single `_debezium_metadata` struct column instead of flattening them to the "
+          + "root level. The change-operation-type column and the log-position column (e.g. the "
+          + "Postgres LSN) are kept at the root level so that payload ordering keeps working. When "
+          + "this property is not set explicitly, the per-database transformer default is used "
+          + "(PostgresDebeziumTransformer defaults to true).");
+
+  public static final ConfigProperty<Boolean> SCHEMA_AS_NULLABLE = ConfigProperty
+      .key(PREFIX + "schema.nullable.enable")
+      .defaultValue(true)
+      .withAlternatives(OLD_PREFIX + "schema.nullable.enable")
+      .markAdvanced()
+      .sinceVersion("1.3.0")
+      .withDocumentation("When enabled, all columns in the transformed Debezium schema are marked "
+          + "as nullable. This keeps the output schema compatible with the nullable columns that "
+          + "Debezium change events produce.");
+}