From a36519856ad1d802146144bf9100e0b12a069317 Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Mon, 29 Jun 2026 15:26:21 -0700 Subject: [PATCH 1/6] feat(utilities): add Postgres Debezium CDC transformer 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). Co-Authored-By: Claude Opus 4.8 --- .../config/DebeziumTransformerConfig.java | 69 +++++ .../debezium/AbstractDebeziumTransformer.java | 242 ++++++++++++++++++ .../debezium/PostgresDebeziumTransformer.java | 82 ++++++ .../debezium/DebeziumTransformerTestBase.java | 69 +++++ .../TestAbstractDebeziumTransformer.java | 189 ++++++++++++++ .../TestPostgresDebeziumTransformer.java | 122 +++++++++ 6 files changed, 773 insertions(+) create mode 100644 hudi-utilities/src/main/java/org/apache/hudi/utilities/config/DebeziumTransformerConfig.java create mode 100644 hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java create mode 100644 hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/PostgresDebeziumTransformer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/DebeziumTransformerTestBase.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/DebeziumTransformerConfig.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/DebeziumTransformerConfig.java new file mode 100644 index 0000000000000..12e4e4fa79ff9 --- /dev/null +++ b/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 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 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."); +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java new file mode 100644 index 0000000000000..a295e044a7864 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java @@ -0,0 +1,242 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.common.util.ConfigUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.utilities.config.DebeziumTransformerConfig; +import org.apache.hudi.utilities.transform.Transformer; + +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Column; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.functions; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_ENABLED; +import static org.apache.hudi.utilities.streamer.BaseErrorTableWriter.ERROR_TABLE_CURRUPT_RECORD_COL_NAME; + +/** + * Base {@link Transformer} that flattens a Debezium change-event envelope into a Hudi row. + * + *

A Debezium change event is a nested record of the form + * {@code {op, ts_ms, before:{...}, after:{...}, source:{...}}}. This transformer: + *

    + *
  • selects the {@code before} image for deletes and the {@code after} image otherwise, + * and explodes it to the row's top level;
    • + *
  • surfaces the common Debezium metadata columns (operation type, processing/origin + * timestamps, shard) along with any database-specific metadata columns supplied by the + * subclass;
    • + *
  • optionally nests the metadata columns under a single {@code _debezium_metadata} struct + * (see {@link DebeziumTransformerConfig#ENABLE_NESTED_FIELDS});
    • + *
  • optionally preserves the error-table corrupt-record column when the error table is + * enabled;
    • + *
  • applies an optional database-specific post-processing step (e.g. ordering/sequence + * columns, LSN defaulting);
    • + *
  • normalizes column nullability (see + * {@link DebeziumTransformerConfig#SCHEMA_AS_NULLABLE}).
    • + *
+ * + *

The flattened column names are defined in {@link DebeziumConstants}; the matching + * {@code DebeziumAvroPayload} implementations rely on these names for merge/ordering semantics. + * + *

Subclasses configure the database-specific behavior purely through the constructor; there is + * no abstract method to implement. + */ +public class AbstractDebeziumTransformer implements Transformer { + + public static final String DEBEZIUM_METADATA_FIELD = "_debezium_metadata"; + private static final String DATA_FIELD = "__data"; + + private static final List DEFAULT_ROOT_LEVEL_METADATA_COLUMNS = Arrays.asList( + new Column(DebeziumConstants.INCOMING_OP_FIELD).alias(DebeziumConstants.FLATTENED_OP_COL_NAME)); + + private static final List DEFAULT_NESTED_METADATA_COLUMNS = Arrays.asList( + new Column(DebeziumConstants.INCOMING_TS_MS_FIELD).alias(DebeziumConstants.UPSTREAM_PROCESSING_TS_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_NAME_FIELD).alias(DebeziumConstants.FLATTENED_SHARD_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_TS_MS_FIELD).alias(DebeziumConstants.FLATTENED_TS_COL_NAME)); + + private final List typeSpecificMetadataColumns; + private final Option, Dataset>> postProcessingOption; + private final boolean nestedFieldsEnabledByDefault; + + protected AbstractDebeziumTransformer( + List typeSpecificMetadataColumns, + Option, Dataset>> postProcessingOption) { + this(typeSpecificMetadataColumns, postProcessingOption, false); + } + + /** + * @param typeSpecificMetadataColumns database-specific metadata columns (already aliased to their + * flattened output names). + * @param postProcessingOption optional post-flatten transformation applied to the result. + * @param nestedFieldsEnabledByDefault default used for + * {@link DebeziumTransformerConfig#ENABLE_NESTED_FIELDS} when + * the property is not set explicitly. Lets a subclass (e.g. + * Postgres) opt into nested metadata by default. + */ + protected AbstractDebeziumTransformer( + List typeSpecificMetadataColumns, + Option, Dataset>> postProcessingOption, + boolean nestedFieldsEnabledByDefault) { + this.typeSpecificMetadataColumns = typeSpecificMetadataColumns; + this.postProcessingOption = postProcessingOption; + this.nestedFieldsEnabledByDefault = nestedFieldsEnabledByDefault; + } + + @Override + public Dataset apply(JavaSparkContext javaSparkContext, SparkSession sparkSession, Dataset rowDataset, TypedProperties props) { + if (rowDataset.columns().length == 0) { + return rowDataset; + } + // Pick selective debezium meta fields: pick the row values from before field for delete record + // and row values from after field for insert or update records. + rowDataset = rowDataset + .withColumn(DATA_FIELD, + functions.when(new Column(DebeziumConstants.INCOMING_OP_FIELD).equalTo(DebeziumConstants.DELETE_OP), + new Column(DebeziumConstants.INCOMING_BEFORE_FIELD)) + .otherwise(new Column(DebeziumConstants.INCOMING_AFTER_FIELD))) + .drop(DebeziumConstants.INCOMING_AFTER_FIELD, DebeziumConstants.INCOMING_BEFORE_FIELD); + + List allColumns = new ArrayList<>(); + boolean enableNestedFields = isNestedFieldsEnabled(props); + + // When nested fields are enabled, only _change_operation_type and root-level metadata columns should be at root level + if (enableNestedFields) { + Column lsnColumn = null; + List otherMetadata = new ArrayList<>(); + + // Extract LSN column to root level, keep other metadata nested + for (Column col : typeSpecificMetadataColumns) { + String colStr = col.toString(); + if (colStr.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME)) { + lsnColumn = col; + } else { + otherMetadata.add(col); + } + } + + List nestedMetadataFields = new ArrayList<>(); + nestedMetadataFields.addAll(DEFAULT_NESTED_METADATA_COLUMNS); + nestedMetadataFields.addAll(otherMetadata); + + // Only add schema field if it exists in the source struct (not all databases have this field) + if (hasSchemaField(rowDataset)) { + nestedMetadataFields.add(new Column(DebeziumConstants.INCOMING_SOURCE_SCHEMA_FIELD).alias(DebeziumConstants.FLATTENED_SCHEMA_NAME)); + } + + rowDataset = rowDataset.withColumn(DEBEZIUM_METADATA_FIELD, + functions.struct(nestedMetadataFields.toArray(new Column[]{}))); + allColumns.add(new Column(DEBEZIUM_METADATA_FIELD)); + + allColumns.addAll(DEFAULT_ROOT_LEVEL_METADATA_COLUMNS); + if (lsnColumn != null) { + // Add LSN column to root level + allColumns.add(lsnColumn); + } + } else { + // When nested fields are disabled, all metadata fields are at root level + allColumns.addAll(DEFAULT_ROOT_LEVEL_METADATA_COLUMNS); + allColumns.addAll(DEFAULT_NESTED_METADATA_COLUMNS); + allColumns.addAll(typeSpecificMetadataColumns); + } + + allColumns.add(new Column(String.format("%s.*", DATA_FIELD))); + + if (ConfigUtils.getBooleanWithAltKeys(props, ERROR_TABLE_ENABLED)) { + if (!Arrays.stream(rowDataset.columns()).collect(Collectors.toList()) + .contains(ERROR_TABLE_CURRUPT_RECORD_COL_NAME)) { + rowDataset = rowDataset.withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME, functions.lit(null)); + } + allColumns.add(new Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME)); + } + + Dataset flattened = rowDataset.select(allColumns.toArray(new Column[]{})); + Dataset debeziumDataset = postProcessingOption.map(postProcessing -> postProcessing.apply(flattened)).orElse(flattened); + + if (ConfigUtils.getBooleanWithAltKeys(props, DebeziumTransformerConfig.SCHEMA_AS_NULLABLE)) { + return convertColumnsToNullable(sparkSession, debeziumDataset); + } + + Set nonNullableColumns = new HashSet<>(); + for (StructField field : rowDataset.schema().fields()) { + if (field.dataType() instanceof StructType && DATA_FIELD.equals(field.name())) { + nonNullableColumns.addAll(Arrays.stream(((StructType) field.dataType()).fields()) + .filter(dataField -> !dataField.nullable()) + .map(StructField::name) + .collect(Collectors.toSet())); + } + } + + // Apply correct nullability to the transformed schema + StructField[] updatedStructFields = Arrays.stream(debeziumDataset.schema().fields()) + .map(field -> field.nullable() && !nonNullableColumns.contains(field.name()) + ? new StructField(field.name(), field.dataType(), true, field.metadata()) + : new StructField(field.name(), field.dataType(), false, field.metadata())) + .toArray(StructField[]::new); + + return sparkSession.createDataFrame(debeziumDataset.rdd(), new StructType(updatedStructFields)); + } + + /** + * Resolves whether to nest the metadata columns. An explicitly set property always wins; when the + * property is absent the per-subclass default ({@link #nestedFieldsEnabledByDefault}) is used. + */ + private boolean isNestedFieldsEnabled(TypedProperties props) { + return ConfigUtils.getRawValueWithAltKeys(props, DebeziumTransformerConfig.ENABLE_NESTED_FIELDS) + .map(value -> Boolean.parseBoolean(value.toString())) + .orElse(nestedFieldsEnabledByDefault); + } + + /** + * Rebuilds the dataset with every column marked nullable. + */ + private static Dataset convertColumnsToNullable(SparkSession sparkSession, Dataset dataset) { + StructField[] modifiedStructFields = Arrays.stream(dataset.schema().fields()) + .map(field -> new StructField(field.name(), field.dataType(), true, field.metadata())) + .toArray(StructField[]::new); + return sparkSession.createDataFrame(dataset.rdd(), new StructType(modifiedStructFields)); + } + + private static boolean hasSchemaField(Dataset rowDataset) { + return Arrays.stream(rowDataset.schema().fields()) + .filter(field -> DebeziumConstants.INCOMING_SOURCE_FIELD.equals(field.name()) && field.dataType() instanceof StructType) + .findFirst() + .map(field -> { + StructType sourceType = (StructType) field.dataType(); + return Arrays.stream(sourceType.fields()) + .anyMatch(sourceField -> "schema".equals(sourceField.name())); + }) + .orElse(false); + } +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/PostgresDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/PostgresDebeziumTransformer.java new file mode 100644 index 0000000000000..45bd5183d95eb --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/PostgresDebeziumTransformer.java @@ -0,0 +1,82 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.sql.Column; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import java.util.Arrays; +import java.util.List; + +import static org.apache.spark.sql.functions.expr; +import static org.apache.spark.sql.functions.when; + +/** + * {@link AbstractDebeziumTransformer} for Postgres Debezium change events. + * + *

Surfaces the Postgres-specific source metadata ({@code txId}, {@code lsn}, {@code xmin}) as the + * flattened {@code _event_tx_id}, {@code _event_lsn} and {@code _event_xmin} columns. The + * {@code _event_lsn} column is the ordering field used by {@code PostgresDebeziumAvroPayload}. + * + *

Post-processing defaults a null {@code _event_lsn} to {@code 0} for snapshot records, since the + * LSN is not populated for rows produced by Debezium incremental snapshots + * (see incremental snapshots). + * + *

Nested metadata is enabled by default for Postgres (the {@code _event_lsn} and operation-type + * columns stay at the root level so payload ordering keeps working); set + * {@code hoodie.streamer.transformer.debezium.nested.fields.enable=false} to flatten all metadata. + */ +public class PostgresDebeziumTransformer extends AbstractDebeziumTransformer { + + // Snapshot operation type emitted by Debezium ("read"). + private static final String SNAPSHOT_OP = "r"; + + private static final List POSTGRES_METADATA = Arrays.asList( + new Column(DebeziumConstants.INCOMING_SOURCE_TXID_FIELD).alias(DebeziumConstants.FLATTENED_TX_ID_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_LSN_FIELD).alias(DebeziumConstants.FLATTENED_LSN_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_XMIN_FIELD).alias(DebeziumConstants.FLATTENED_XMIN_COL_NAME)); + + public PostgresDebeziumTransformer() { + super(POSTGRES_METADATA, Option.of(PostgresDebeziumTransformer::useDefaultValuesForLsnIfNull), true); + } + + /** + * Defaults a null {@code _event_lsn} to {@code 0} for snapshot records. The LSN is null when a + * table is added via a Debezium incremental snapshot; leaving it null would break LSN-based + * ordering in the payload. + * + * @param dataset flattened Postgres Debezium dataset. + * @return dataset where null {@code _event_lsn} values on snapshot rows are replaced with 0. + */ + private static Dataset useDefaultValuesForLsnIfNull(Dataset dataset) { + if (!Arrays.asList(dataset.columns()).contains(DebeziumConstants.FLATTENED_LSN_COL_NAME)) { + return dataset; + } + + return dataset.withColumn(DebeziumConstants.FLATTENED_LSN_COL_NAME, when( + dataset.col(DebeziumConstants.FLATTENED_OP_COL_NAME).equalTo(SNAPSHOT_OP), + expr(String.format("coalesce(%s, cast(0 as long))", DebeziumConstants.FLATTENED_LSN_COL_NAME)) + ).otherwise(dataset.col(DebeziumConstants.FLATTENED_LSN_COL_NAME))); + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/DebeziumTransformerTestBase.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/DebeziumTransformerTestBase.java new file mode 100644 index 0000000000000..b1ba1cc02aad7 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/DebeziumTransformerTestBase.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.transform.debezium; + +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; + +import java.io.IOException; +import java.util.Arrays; + +import static org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest; + +/** + * Lightweight Spark test base for the Debezium transformer tests. Spins up a single local + * SparkSession for the test class and parses inline JSON fixtures into {@link Dataset}s that mimic + * Debezium change-event envelopes. + */ +public abstract class DebeziumTransformerTestBase { + + protected static SparkSession spark; + protected static JavaSparkContext jsc; + + @BeforeAll + static void setUpSpark() throws IOException { + spark = SparkSession.builder() + .config(getSparkConfForTest(DebeziumTransformerTestBase.class.getName())) + .getOrCreate(); + jsc = JavaSparkContext.fromSparkContext(spark.sparkContext()); + } + + @AfterAll + static void tearDownSpark() throws IOException { + if (spark != null) { + spark.close(); + spark = null; + jsc = null; + } + } + + /** + * Parses each argument as a single JSON document into a {@link Dataset}. Passing several + * envelopes together lets Spark infer struct types for {@code before}/{@code after} even when an + * individual envelope leaves one of them null (inserts have no {@code before}, deletes have no + * {@code after}). + */ + protected Dataset jsonToDataset(String... jsonDocs) { + return spark.read().json(jsc.parallelize(Arrays.asList(jsonDocs), 1)); + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java new file mode 100644 index 0000000000000..995a17fc1b209 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java @@ -0,0 +1,189 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.utilities.config.DebeziumTransformerConfig; + +import org.apache.spark.sql.Column; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.function.Function; + +import static org.apache.hudi.utilities.transform.debezium.AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the database-agnostic flattening behavior of {@link AbstractDebeziumTransformer}. + */ +class TestAbstractDebeziumTransformer extends DebeziumTransformerTestBase { + + // One insert (op=c), one update (op=u), one delete (op=d). before/after differ so the test can + // prove which image is selected. The delete carries values only in `before` (after is null). + private static final String INSERT_EVENT = + "{\"op\":\"c\",\"ts_ms\":1700000000500," + + "\"before\":null," + + "\"after\":{\"id\":1,\"name\":\"Alice\",\"email\":\"alice@x.com\"}," + + "\"source\":{\"name\":\"pgdb\",\"ts_ms\":1700000000000,\"schema\":\"public\"}}"; + private static final String UPDATE_EVENT = + "{\"op\":\"u\",\"ts_ms\":1700000001500," + + "\"before\":{\"id\":2,\"name\":\"Bob\",\"email\":\"old@x.com\"}," + + "\"after\":{\"id\":2,\"name\":\"Bob\",\"email\":\"new@x.com\"}," + + "\"source\":{\"name\":\"pgdb\",\"ts_ms\":1700000001000,\"schema\":\"public\"}}"; + private static final String DELETE_EVENT = + "{\"op\":\"d\",\"ts_ms\":1700000002500," + + "\"before\":{\"id\":3,\"name\":\"Carol\",\"email\":\"carol@x.com\"}," + + "\"after\":null," + + "\"source\":{\"name\":\"pgdb\",\"ts_ms\":1700000002000,\"schema\":\"public\"}}"; + + @Test + void testAfterUsedForInsertUpdate_beforeUsedForDelete() { + Dataset input = jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT); + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, input, new TypedProperties()); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.containsAll(Arrays.asList("id", "name", "email")), "data columns lifted to root"); + assertFalse(columns.contains(DebeziumConstants.INCOMING_AFTER_FIELD), "after dropped"); + assertFalse(columns.contains(DebeziumConstants.INCOMING_BEFORE_FIELD), "before dropped"); + + List rows = result.orderBy("id").collectAsList(); + assertEquals(3, rows.size()); + + // insert -> taken from `after` + assertEquals("c", rows.get(0).getAs(DebeziumConstants.FLATTENED_OP_COL_NAME)); + assertEquals("Alice", rows.get(0).getAs("name")); + assertEquals("alice@x.com", rows.get(0).getAs("email")); + + // update -> taken from `after` (the NEW email, not the before/old one) + assertEquals("u", rows.get(1).getAs(DebeziumConstants.FLATTENED_OP_COL_NAME)); + assertEquals("new@x.com", rows.get(1).getAs("email")); + + // delete -> taken from `before` (after is null, but we keep the key/row) + assertEquals("d", rows.get(2).getAs(DebeziumConstants.FLATTENED_OP_COL_NAME)); + assertEquals("Carol", rows.get(2).getAs("name")); + assertEquals("carol@x.com", rows.get(2).getAs("email")); + } + + @Test + void testDefaultMetadataColumnsAtRootWhenFlat() { + Dataset input = jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT); + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, input, new TypedProperties()); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME)); + assertTrue(columns.contains(DebeziumConstants.UPSTREAM_PROCESSING_TS_COL_NAME)); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_SHARD_NAME)); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_TS_COL_NAME)); + assertFalse(columns.contains(DEBEZIUM_METADATA_FIELD), "no nested struct when flat"); + + // shard name comes from source.name + assertEquals("pgdb", result.orderBy("id").collectAsList().get(0).getAs(DebeziumConstants.FLATTENED_SHARD_NAME)); + } + + @Test + void testNestedModeGroupsMetadataButKeepsOpAndLsnAtRoot() { + // Supply an LSN-named metadata column so the LSN-to-root branch is exercised. + Column lsnColumn = new Column(DebeziumConstants.INCOMING_SOURCE_NAME_FIELD).alias(DebeziumConstants.FLATTENED_LSN_COL_NAME); + TestableDebeziumTransformer transformer = + new TestableDebeziumTransformer(Collections.singletonList(lsnColumn), Option.empty(), false); + + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "true"); + + Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), props); + List columns = Arrays.asList(result.columns()); + + assertTrue(columns.contains(DEBEZIUM_METADATA_FIELD), "metadata grouped under a struct"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op stays at root"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "LSN stays at root"); + assertFalse(columns.contains(DebeziumConstants.FLATTENED_SHARD_NAME), "shard moved into the nested struct"); + + Row metadata = result.first().getAs(DEBEZIUM_METADATA_FIELD); + List nested = Arrays.asList(metadata.schema().fieldNames()); + assertTrue(nested.contains(DebeziumConstants.FLATTENED_SHARD_NAME)); + assertTrue(nested.contains(DebeziumConstants.FLATTENED_SCHEMA_NAME), "source.schema present -> nested schema col added"); + assertFalse(nested.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "LSN is at root, not nested"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testNestedDefaultIsUsedWhenPropertyAbsent(boolean subclassDefault) { + TestableDebeziumTransformer transformer = + new TestableDebeziumTransformer(Collections.emptyList(), Option.empty(), subclassDefault); + + // property NOT set -> subclass default decides + Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), new TypedProperties()); + assertEquals(subclassDefault, Arrays.asList(result.columns()).contains(DEBEZIUM_METADATA_FIELD)); + } + + @Test + void testExplicitPropertyOverridesSubclassDefault() { + // subclass defaults to nested=true, but an explicit false must win + TestableDebeziumTransformer transformer = + new TestableDebeziumTransformer(Collections.emptyList(), Option.empty(), true); + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); + + Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), props); + assertFalse(Arrays.asList(result.columns()).contains(DEBEZIUM_METADATA_FIELD)); + } + + @Test + void testPostProcessingHookIsApplied() { + Function, Dataset> addColumn = + ds -> ds.withColumn("processed", org.apache.spark.sql.functions.lit(true)); + TestableDebeziumTransformer transformer = + new TestableDebeziumTransformer(Collections.emptyList(), Option.of(addColumn), false); + + Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), new TypedProperties()); + assertTrue(Arrays.asList(result.columns()).contains("processed")); + assertTrue(result.collectAsList().stream().allMatch(r -> Boolean.TRUE.equals(r.getAs("processed")))); + } + + @Test + void testEmptyDatasetIsReturnedUnchanged() { + Dataset empty = spark.emptyDataFrame(); + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, empty, new TypedProperties()); + assertEquals(0, result.columns().length); + } + + /** Minimal concrete subclass to exercise the otherwise database-agnostic base. */ + private static class TestableDebeziumTransformer extends AbstractDebeziumTransformer { + TestableDebeziumTransformer() { + super(Collections.emptyList(), Option.empty()); + } + + TestableDebeziumTransformer(List typeSpecificMetadataColumns, + Option, Dataset>> postProcessingOption, + boolean nestedFieldsEnabledByDefault) { + super(typeSpecificMetadataColumns, postProcessingOption, nestedFieldsEnabledByDefault); + } + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java new file mode 100644 index 0000000000000..9d7fd680b38c9 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java @@ -0,0 +1,122 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.utilities.config.DebeziumTransformerConfig; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the Postgres-specific behavior of {@link PostgresDebeziumTransformer}: the surfaced source + * metadata columns, the nested-by-default layout, and the snapshot LSN defaulting. + */ +class TestPostgresDebeziumTransformer extends DebeziumTransformerTestBase { + + // A Postgres source struct carries txId/lsn/xmin/schema in addition to name/ts_ms. + // before is populated (same shape as after) so Spark infers it as a struct; these tests exercise + // inserts/snapshots (op c/r), which take `after` regardless of before's contents. + private static String pgEvent(String op, long id, long lsn, boolean nullLsn) { + String row = "{\"id\":" + id + ",\"name\":\"row" + id + "\",\"email\":\"e" + id + "@x.com\"}"; + return "{\"op\":\"" + op + "\",\"ts_ms\":1700000000500," + + "\"before\":" + row + "," + + "\"after\":" + row + "," + + "\"source\":{\"name\":\"pgdb\",\"ts_ms\":1700000000000,\"schema\":\"public\"," + + "\"txId\":" + (500 + id) + ",\"lsn\":" + (nullLsn ? "null" : lsn) + ",\"xmin\":" + (9000 + id) + "}}"; + } + + @Test + void testPostgresMetadataColumnsAreSurfaced() { + TypedProperties props = new TypedProperties(); + // force flat so all metadata columns sit at root and are easy to assert + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); + + Dataset result = new PostgresDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(pgEvent("c", 1, 1001, false)), props); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_TX_ID_COL_NAME), "_event_tx_id present"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "_event_lsn present"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_XMIN_COL_NAME), "_event_xmin present"); + + Row row = result.first(); + long lsn = row.getAs(DebeziumConstants.FLATTENED_LSN_COL_NAME); + long txId = row.getAs(DebeziumConstants.FLATTENED_TX_ID_COL_NAME); + assertEquals(1001L, lsn); + assertEquals(501L, txId); + } + + @Test + void testPostgresDefaultsToNestedMetadata() { + // No nested property set -> Postgres should nest by default, with op + LSN kept at root. + Dataset result = new PostgresDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(pgEvent("c", 1, 1001, false)), new TypedProperties()); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), "nested by default for Postgres"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op at root"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "LSN at root (payload ordering field)"); + assertFalse(columns.contains(DebeziumConstants.FLATTENED_TX_ID_COL_NAME), "tx id is nested, not at root"); + + Row metadata = result.first().getAs(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD); + assertTrue(Arrays.asList(metadata.schema().fieldNames()).contains(DebeziumConstants.FLATTENED_TX_ID_COL_NAME)); + } + + @Test + void testNullLsnDefaultedToZeroForSnapshotRows() { + // Snapshot rows (op = r) from incremental snapshots can have a null LSN; it must become 0 + // so the payload's "higher LSN wins" ordering does not choke on null. + Dataset input = jsonToDataset( + pgEvent("r", 1, 0, true), // snapshot, null lsn -> expect 0 + pgEvent("r", 2, 2002, false) // snapshot, real lsn -> unchanged + ); + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); + + Dataset result = new PostgresDebeziumTransformer().apply(jsc, spark, input, props); + List rows = result.orderBy("id").collectAsList(); + + long defaultedLsn = rows.get(0).getAs(DebeziumConstants.FLATTENED_LSN_COL_NAME); + long preservedLsn = rows.get(1).getAs(DebeziumConstants.FLATTENED_LSN_COL_NAME); + assertEquals(0L, defaultedLsn, "null snapshot LSN defaulted to 0"); + assertEquals(2002L, preservedLsn, "real LSN preserved"); + } + + @Test + void testNullLsnNotDefaultedForNonSnapshotRows() { + // For a real change event (op != r) a null LSN is left as null (only snapshots are defaulted). + Dataset input = jsonToDataset(pgEvent("c", 1, 0, true)); + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); + + Dataset result = new PostgresDebeziumTransformer().apply(jsc, spark, input, props); + assertNull(result.first().getAs(DebeziumConstants.FLATTENED_LSN_COL_NAME)); + } +} From b005c3bfa3c2135b3adebdb5cdaae771c391f59a Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Mon, 29 Jun 2026 20:20:06 -0700 Subject: [PATCH 2/6] feat(utilities): add Mysql Debezium CDC transformer 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 (".") consumed by the existing MySqlDebeziumAvroPayload. Metadata flat by default; nested layout supported via the shared config. Adds unit tests (flat + nested + seq). Co-Authored-By: Claude Opus 4.8 --- .../debezium/MysqlDebeziumTransformer.java | 82 +++++++++++++ .../TestMysqlDebeziumTransformer.java | 110 ++++++++++++++++++ 2 files changed, 192 insertions(+) create mode 100644 hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java new file mode 100644 index 0000000000000..663dc80f599e1 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java @@ -0,0 +1,82 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.common.util.Option; + +import org.apache.spark.sql.Column; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.functions; + +import java.util.Arrays; +import java.util.List; + +/** + * {@link AbstractDebeziumTransformer} for MySQL Debezium change events. + * + *

Surfaces the MySQL binlog coordinates ({@code file}, {@code pos}, {@code row}) as the flattened + * {@code _event_bin_file}, {@code _event_pos} and {@code _event_row} columns, and derives the + * {@code _event_seq} ordering column as {@code "."} (e.g. {@code "000001.100"} + * for a binlog file {@code "mysql-bin.000001"} at position {@code 100}). {@code _event_seq} is the + * precombine/ordering field consumed by {@code MySqlDebeziumAvroPayload}. + * + *

Metadata is flattened to the root level by default; set + * {@code hoodie.streamer.transformer.debezium.nested.fields.enable=true} to group it under a + * {@code _debezium_metadata} struct instead. + */ +public class MysqlDebeziumTransformer extends AbstractDebeziumTransformer { + + private static final List MYSQL_METADATA = Arrays.asList( + new Column(DebeziumConstants.INCOMING_SOURCE_FILE_FIELD).alias(DebeziumConstants.FLATTENED_FILE_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_POS_FIELD).alias(DebeziumConstants.FLATTENED_POS_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_ROW_FIELD).alias(DebeziumConstants.FLATTENED_ROW_COL_NAME)); + + public MysqlDebeziumTransformer() { + super(MYSQL_METADATA, Option.of(MysqlDebeziumTransformer::applySeqNo)); + } + + /** + * Builds the {@code _event_seq} ordering column from the binlog file and position. The file column + * holds a name like {@code "mysql-bin.000001"}; only the numeric suffix after the last dot is used, + * yielding a sequence such as {@code "000001.100"}. Handles both the flat and nested metadata + * layouts (reading {@code file}/{@code pos} from the {@code _debezium_metadata} struct when nested). + * + * @param dataset flattened MySQL Debezium dataset. + * @return dataset with the {@code _event_seq} column added. + */ + private static Dataset applySeqNo(Dataset dataset) { + boolean isNested = Arrays.asList(dataset.columns()).contains(DEBEZIUM_METADATA_FIELD); + + Column fileCol = isNested + ? dataset.col(DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_FILE_COL_NAME) + : dataset.col(DebeziumConstants.FLATTENED_FILE_COL_NAME); + + Column posCol = isNested + ? dataset.col(DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_POS_COL_NAME) + : dataset.col(DebeziumConstants.FLATTENED_POS_COL_NAME); + + return dataset.withColumn(DebeziumConstants.ADDED_SEQ_COL_NAME, functions.concat( + functions.substring_index(fileCol, ".", -1), + functions.lit("."), + posCol)); + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java new file mode 100644 index 0000000000000..93f3a4134a7cc --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java @@ -0,0 +1,110 @@ +/* + * 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.transform.debezium; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.debezium.DebeziumConstants; +import org.apache.hudi.utilities.config.DebeziumTransformerConfig; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the MySQL-specific behavior of {@link MysqlDebeziumTransformer}: the surfaced binlog + * metadata columns and the derived {@code _event_seq} ordering column, in both flat and nested + * layouts. + */ +class TestMysqlDebeziumTransformer extends DebeziumTransformerTestBase { + + // A MySQL source struct carries binlog file/pos/row in addition to name/ts_ms. before is populated + // (same shape as after) so Spark infers it as a struct; these tests use inserts (op=c) which take + // `after` regardless of before's contents. + private static String mysqlEvent(String op, long id, String file, long pos) { + String row = "{\"id\":" + id + ",\"title\":\"t" + id + "\"}"; + return "{\"op\":\"" + op + "\",\"ts_ms\":1700000000500," + + "\"before\":" + row + "," + + "\"after\":" + row + "," + + "\"source\":{\"name\":\"mysqldb\",\"ts_ms\":1700000000000," + + "\"file\":\"" + file + "\",\"pos\":" + pos + ",\"row\":0}}"; + } + + @Test + void testMysqlMetadataColumnsAndSeqSurfacedFlat() { + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); + + Dataset result = new MysqlDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(mysqlEvent("c", 1, "mysql-bin.000001", 100)), props); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME), "_event_bin_file present"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_POS_COL_NAME), "_event_pos present"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_ROW_COL_NAME), "_event_row present"); + assertTrue(columns.contains(DebeziumConstants.ADDED_SEQ_COL_NAME), "_event_seq present"); + + // _event_seq = "." -> "000001.100" + String seq = result.first().getAs(DebeziumConstants.ADDED_SEQ_COL_NAME); + assertEquals("000001.100", seq); + } + + @Test + void testFlatByDefault() { + // MySQL does not opt into nested metadata, so with no property set the layout is flat. + Dataset result = new MysqlDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(mysqlEvent("c", 1, "mysql-bin.000001", 100)), new TypedProperties()); + + List columns = Arrays.asList(result.columns()); + assertFalse(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), + "MySQL should be flat by default"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME), "_event_bin_file at root"); + } + + @Test + void testNestedModeGroupsMetadataButSeqStaysCorrect() { + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "true"); + + Dataset result = new MysqlDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(mysqlEvent("c", 1, "mysql-bin.000001", 100)), props); + + List columns = Arrays.asList(result.columns()); + assertTrue(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), "metadata nested"); + assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op at root"); + assertTrue(columns.contains(DebeziumConstants.ADDED_SEQ_COL_NAME), "_event_seq at root"); + assertFalse(columns.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME), "binlog file is nested, not root"); + + Row metadata = result.first().getAs(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD); + List nested = Arrays.asList(metadata.schema().fieldNames()); + assertTrue(nested.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME)); + assertTrue(nested.contains(DebeziumConstants.FLATTENED_POS_COL_NAME)); + assertTrue(nested.contains(DebeziumConstants.FLATTENED_ROW_COL_NAME)); + + // _event_seq must still be computed correctly from the nested file/pos + String seq = result.first().getAs(DebeziumConstants.ADDED_SEQ_COL_NAME); + assertEquals("000001.100", seq); + } +} From 1562bd91da1830e253e7b561a015ecb9751a33c3 Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Tue, 30 Jun 2026 08:51:41 -0700 Subject: [PATCH 3/6] feat(utilities): nest Debezium ordering fields when nested metadata is 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. Co-Authored-By: Claude Opus 4.8 --- .../model/debezium/DebeziumConstants.java | 3 ++ .../hudi/common/table/HoodieTableConfig.java | 48 +++++++++++++++++-- .../common/table/TestHoodieTableConfig.java | 31 ++++++++++++ .../utilities/streamer/HoodieStreamer.java | 4 +- .../debezium/AbstractDebeziumTransformer.java | 2 +- 5 files changed, 82 insertions(+), 6 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/debezium/DebeziumConstants.java b/hudi-common/src/main/java/org/apache/hudi/common/model/debezium/DebeziumConstants.java index 3acaca5137555..a8fecf9387377 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/model/debezium/DebeziumConstants.java +++ b/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 META_COLUMNS = Collections.unmodifiableList(Arrays.asList( FLATTENED_OP_COL_NAME, diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java index 6573ceed6b8ea..d173f2bc3d8dc 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java @@ -849,6 +849,16 @@ public static Map inferMergingConfigsForV9TableCreation(RecordMe String recordMergeStrategyId, String orderingFieldName, HoodieTableVersion tableVersion) { + return inferMergingConfigsForV9TableCreation(recordMergeMode, payloadClassName, recordMergeStrategyId, + orderingFieldName, tableVersion, false); + } + + public static Map inferMergingConfigsForV9TableCreation(RecordMergeMode recordMergeMode, + String payloadClassName, + String recordMergeStrategyId, + String orderingFieldName, + HoodieTableVersion tableVersion, + boolean nestedDebeziumMetadataEnabled) { Map 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 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 reconciledConfigs) { + private static void handlePayloadAdhocConfigs(String payloadClassName, Map 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 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 inferMergingConfigsForWrites(RecordMergeMode recordMergeMode, + String payloadClassName, + String recordMergeStrategyId, + String orderingFieldNamesAsString, + HoodieTableVersion tableVersion, + boolean nestedDebeziumMetadataEnabled) { if (tableVersion.greaterThanOrEquals(HoodieTableVersion.NINE)) { Map 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())), diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java index 9d09629017c12..f7a84ae8a1ef1 100644 --- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java +++ b/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 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 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 pgFlat = HoodieTableConfig.inferMergingConfigsForV9TableCreation( + null, pgPayload, null, "ts", HoodieTableVersion.NINE, false); + Map 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 diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java index 9fc9dbff66a9d..01f74fa61d082 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java @@ -66,6 +66,7 @@ import org.apache.hudi.utilities.IdentitySplitter; import org.apache.hudi.utilities.UtilHelpers; import org.apache.hudi.utilities.checkpointing.InitialCheckPointProvider; +import org.apache.hudi.utilities.config.DebeziumTransformerConfig; import org.apache.hudi.utilities.ingestion.HoodieIngestionException; import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics; import org.apache.hudi.utilities.ingestion.HoodieIngestionService; @@ -158,7 +159,8 @@ public HoodieStreamer(Config cfg, JavaSparkContext jssc, FileSystem fs, Configur Triple mergingConfigs = HoodieTableConfig.inferMergingConfigsForWrites( cfg.recordMergeMode, cfg.payloadClassName, cfg.recordMergeStrategyId, cfg.sourceOrderingFields, - HoodieTableVersion.fromVersionCode(ConfigUtils.getIntWithAltKeys(this.properties, HoodieWriteConfig.WRITE_TABLE_VERSION))); + HoodieTableVersion.fromVersionCode(ConfigUtils.getIntWithAltKeys(this.properties, HoodieWriteConfig.WRITE_TABLE_VERSION)), + ConfigUtils.getBooleanWithAltKeys(this.properties, DebeziumTransformerConfig.ENABLE_NESTED_FIELDS)); cfg.recordMergeMode = mergingConfigs.getLeft(); cfg.recordMergeStrategyId = mergingConfigs.getRight(); if (cfg.initialCheckpointProvider != null && cfg.checkpoint == null) { diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java index a295e044a7864..50709d0d8836a 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java @@ -75,7 +75,7 @@ */ public class AbstractDebeziumTransformer implements Transformer { - public static final String DEBEZIUM_METADATA_FIELD = "_debezium_metadata"; + public static final String DEBEZIUM_METADATA_FIELD = DebeziumConstants.DEBEZIUM_METADATA_FIELD; private static final String DATA_FIELD = "__data"; private static final List DEFAULT_ROOT_LEVEL_METADATA_COLUMNS = Arrays.asList( From b02ff5426cb609142a6e7e657bdeb66b45bdc2e1 Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Tue, 30 Jun 2026 15:37:05 -0700 Subject: [PATCH 4/6] test(utilities): cover error-table passthrough and non-nullable schema 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). Co-Authored-By: Claude Opus 4.8 --- ...TestHoodieTableConfigDebeziumOrdering.java | 65 +++++++++++++++++ .../TestAbstractDebeziumTransformer.java | 71 +++++++++++++++++++ 2 files changed, 136 insertions(+) create mode 100644 hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java new file mode 100644 index 0000000000000..479be82be3994 --- /dev/null +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java @@ -0,0 +1,65 @@ +/* + * 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.model.debezium.DebeziumConstants; +import org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload; +import org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload; + +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 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 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 flat = HoodieTableConfig.inferMergingConfigsForV9TableCreation( + null, PostgresDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE, false); + Map 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())); + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java index 995a17fc1b209..e50303a1dd5e1 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java @@ -26,6 +26,11 @@ import org.apache.spark.sql.Column; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.functions; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructType; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -35,9 +40,12 @@ import java.util.List; import java.util.function.Function; +import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_ENABLED; +import static org.apache.hudi.utilities.streamer.BaseErrorTableWriter.ERROR_TABLE_CURRUPT_RECORD_COL_NAME; import static org.apache.hudi.utilities.transform.debezium.AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -174,6 +182,69 @@ void testEmptyDatasetIsReturnedUnchanged() { assertEquals(0, result.columns().length); } + @Test + void testSchemaAsNullableFalsePreservesNonNullableSourceFields() { + // "id" is declared non-nullable in the source row schema; "name" is nullable. + StructType rowDataSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("id", DataTypes.LongType, false, Metadata.empty()), + DataTypes.createStructField("name", DataTypes.StringType, true, Metadata.empty()))); + StructType sourceSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("name", DataTypes.StringType, true, Metadata.empty()), + DataTypes.createStructField("ts_ms", DataTypes.LongType, true, Metadata.empty()))); + StructType envelopeSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("op", DataTypes.StringType, true, Metadata.empty()), + DataTypes.createStructField("ts_ms", DataTypes.LongType, true, Metadata.empty()), + DataTypes.createStructField("before", rowDataSchema, true, Metadata.empty()), + DataTypes.createStructField("after", rowDataSchema, true, Metadata.empty()), + DataTypes.createStructField("source", sourceSchema, true, Metadata.empty()))); + + Row afterRow = RowFactory.create(1L, "Alice"); + Row sourceRow = RowFactory.create("pgdb", 1700000000000L); + Row envelopeRow = RowFactory.create("c", 1700000000500L, null, afterRow, sourceRow); + Dataset input = spark.createDataFrame(Collections.singletonList(envelopeRow), envelopeSchema); + + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.SCHEMA_AS_NULLABLE.key(), "false"); + + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, input, props); + + assertFalse(result.schema().apply("id").nullable(), "id was non-nullable in the source schema"); + assertTrue(result.schema().apply("name").nullable(), "name was nullable in the source schema"); + long id = result.first().getAs("id"); + assertEquals(1L, id); + assertEquals("Alice", result.first().getAs("name")); + } + + @Test + void testErrorTableEnabledAddsNullCorruptRecordWhenMissing() { + // Include UPDATE_EVENT alongside so Spark's JSON inference sees a non-null `before` + // example and types the column as a struct rather than (incorrectly) as a string. + TypedProperties props = new TypedProperties(); + props.setProperty(ERROR_TABLE_ENABLED.key(), "true"); + + Dataset result = new TestableDebeziumTransformer() + .apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT), props); + + assertTrue(Arrays.asList(result.columns()).contains(ERROR_TABLE_CURRUPT_RECORD_COL_NAME), + "_corrupt_record column added when error table is enabled"); + Row insertRow = result.where(new Column(DebeziumConstants.FLATTENED_OP_COL_NAME).equalTo("c")).first(); + assertNull(insertRow.getAs(ERROR_TABLE_CURRUPT_RECORD_COL_NAME), "no corrupt value was present on input"); + } + + @Test + void testErrorTableEnabledPreservesExistingCorruptRecordValue() { + Dataset input = jsonToDataset(INSERT_EVENT, UPDATE_EVENT) + .withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME, functions.lit("malformed-row")); + TypedProperties props = new TypedProperties(); + props.setProperty(ERROR_TABLE_ENABLED.key(), "true"); + + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, input, props); + + Row insertRow = result.where(new Column(DebeziumConstants.FLATTENED_OP_COL_NAME).equalTo("c")).first(); + assertEquals("malformed-row", insertRow.getAs(ERROR_TABLE_CURRUPT_RECORD_COL_NAME), + "existing corrupt-record value is preserved, not overwritten"); + } + /** Minimal concrete subclass to exercise the otherwise database-agnostic base. */ private static class TestableDebeziumTransformer extends AbstractDebeziumTransformer { TestableDebeziumTransformer() { From 9dea65157c61b05404ef428e36562833da274b20 Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Tue, 30 Jun 2026 18:12:06 -0700 Subject: [PATCH 5/6] test(common): cover the 5-arg backward-compatible ordering-inference 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. Co-Authored-By: Claude Opus 4.8 --- ...TestHoodieTableConfigDebeziumOrdering.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java index 479be82be3994..cd0d9587e72b5 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfigDebeziumOrdering.java @@ -18,9 +18,11 @@ 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; @@ -62,4 +64,26 @@ void postgresOrderingFieldStaysAtRootRegardlessOfNesting() { 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 viaFiveArg = HoodieTableConfig.inferMergingConfigsForV9TableCreation( + null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE); + Map 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 writesViaFiveArg = HoodieTableConfig.inferMergingConfigsForWrites( + null, MySqlDebeziumAvroPayload.class.getName(), null, "ts", HoodieTableVersion.NINE); + Triple 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()); + } } From 4c457806d9bfea0a35377a83c8d8c666d40e1f33 Mon Sep 17 00:00:00 2001 From: Rahil Chertara Date: Wed, 1 Jul 2026 12:29:34 -0700 Subject: [PATCH 6/6] fix(utilities): correct Debezium nullability rebuild and drop redundant 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. --- .../debezium/AbstractDebeziumTransformer.java | 14 +++--- .../debezium/MysqlDebeziumTransformer.java | 6 +-- .../TestAbstractDebeziumTransformer.java | 45 ++++++++++++++++--- .../TestMysqlDebeziumTransformer.java | 6 +-- .../TestPostgresDebeziumTransformer.java | 4 +- 5 files changed, 54 insertions(+), 21 deletions(-) diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java index 50709d0d8836a..9d4b323643968 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java @@ -75,7 +75,6 @@ */ public class AbstractDebeziumTransformer implements Transformer { - public static final String DEBEZIUM_METADATA_FIELD = DebeziumConstants.DEBEZIUM_METADATA_FIELD; private static final String DATA_FIELD = "__data"; private static final List DEFAULT_ROOT_LEVEL_METADATA_COLUMNS = Arrays.asList( @@ -155,9 +154,9 @@ public Dataset apply(JavaSparkContext javaSparkContext, SparkSession sparkS nestedMetadataFields.add(new Column(DebeziumConstants.INCOMING_SOURCE_SCHEMA_FIELD).alias(DebeziumConstants.FLATTENED_SCHEMA_NAME)); } - rowDataset = rowDataset.withColumn(DEBEZIUM_METADATA_FIELD, + rowDataset = rowDataset.withColumn(DebeziumConstants.DEBEZIUM_METADATA_FIELD, functions.struct(nestedMetadataFields.toArray(new Column[]{}))); - allColumns.add(new Column(DEBEZIUM_METADATA_FIELD)); + allColumns.add(new Column(DebeziumConstants.DEBEZIUM_METADATA_FIELD)); allColumns.addAll(DEFAULT_ROOT_LEVEL_METADATA_COLUMNS); if (lsnColumn != null) { @@ -198,11 +197,12 @@ public Dataset apply(JavaSparkContext javaSparkContext, SparkSession sparkS } } - // Apply correct nullability to the transformed schema + // Apply correct nullability to the transformed schema: a column stays non-nullable only if it + // was a non-nullable source column; every other column (including Debezium metadata columns) is nullable. StructField[] updatedStructFields = Arrays.stream(debeziumDataset.schema().fields()) - .map(field -> field.nullable() && !nonNullableColumns.contains(field.name()) - ? new StructField(field.name(), field.dataType(), true, field.metadata()) - : new StructField(field.name(), field.dataType(), false, field.metadata())) + .map(field -> nonNullableColumns.contains(field.name()) + ? new StructField(field.name(), field.dataType(), false, field.metadata()) + : new StructField(field.name(), field.dataType(), true, field.metadata())) .toArray(StructField[]::new); return sparkSession.createDataFrame(debeziumDataset.rdd(), new StructType(updatedStructFields)); diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java index 663dc80f599e1..3c986d32e7d9b 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java @@ -64,14 +64,14 @@ public MysqlDebeziumTransformer() { * @return dataset with the {@code _event_seq} column added. */ private static Dataset applySeqNo(Dataset dataset) { - boolean isNested = Arrays.asList(dataset.columns()).contains(DEBEZIUM_METADATA_FIELD); + boolean isNested = Arrays.asList(dataset.columns()).contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD); Column fileCol = isNested - ? dataset.col(DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_FILE_COL_NAME) + ? dataset.col(DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_FILE_COL_NAME) : dataset.col(DebeziumConstants.FLATTENED_FILE_COL_NAME); Column posCol = isNested - ? dataset.col(DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_POS_COL_NAME) + ? dataset.col(DebeziumConstants.DEBEZIUM_METADATA_FIELD + "." + DebeziumConstants.FLATTENED_POS_COL_NAME) : dataset.col(DebeziumConstants.FLATTENED_POS_COL_NAME); return dataset.withColumn(DebeziumConstants.ADDED_SEQ_COL_NAME, functions.concat( diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java index e50303a1dd5e1..ec07054e230d5 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestAbstractDebeziumTransformer.java @@ -42,7 +42,6 @@ import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_ENABLED; import static org.apache.hudi.utilities.streamer.BaseErrorTableWriter.ERROR_TABLE_CURRUPT_RECORD_COL_NAME; -import static org.apache.hudi.utilities.transform.debezium.AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; @@ -109,7 +108,7 @@ void testDefaultMetadataColumnsAtRootWhenFlat() { assertTrue(columns.contains(DebeziumConstants.UPSTREAM_PROCESSING_TS_COL_NAME)); assertTrue(columns.contains(DebeziumConstants.FLATTENED_SHARD_NAME)); assertTrue(columns.contains(DebeziumConstants.FLATTENED_TS_COL_NAME)); - assertFalse(columns.contains(DEBEZIUM_METADATA_FIELD), "no nested struct when flat"); + assertFalse(columns.contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD), "no nested struct when flat"); // shard name comes from source.name assertEquals("pgdb", result.orderBy("id").collectAsList().get(0).getAs(DebeziumConstants.FLATTENED_SHARD_NAME)); @@ -128,12 +127,12 @@ void testNestedModeGroupsMetadataButKeepsOpAndLsnAtRoot() { Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), props); List columns = Arrays.asList(result.columns()); - assertTrue(columns.contains(DEBEZIUM_METADATA_FIELD), "metadata grouped under a struct"); + assertTrue(columns.contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD), "metadata grouped under a struct"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op stays at root"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "LSN stays at root"); assertFalse(columns.contains(DebeziumConstants.FLATTENED_SHARD_NAME), "shard moved into the nested struct"); - Row metadata = result.first().getAs(DEBEZIUM_METADATA_FIELD); + Row metadata = result.first().getAs(DebeziumConstants.DEBEZIUM_METADATA_FIELD); List nested = Arrays.asList(metadata.schema().fieldNames()); assertTrue(nested.contains(DebeziumConstants.FLATTENED_SHARD_NAME)); assertTrue(nested.contains(DebeziumConstants.FLATTENED_SCHEMA_NAME), "source.schema present -> nested schema col added"); @@ -148,7 +147,7 @@ void testNestedDefaultIsUsedWhenPropertyAbsent(boolean subclassDefault) { // property NOT set -> subclass default decides Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), new TypedProperties()); - assertEquals(subclassDefault, Arrays.asList(result.columns()).contains(DEBEZIUM_METADATA_FIELD)); + assertEquals(subclassDefault, Arrays.asList(result.columns()).contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD)); } @Test @@ -160,7 +159,7 @@ void testExplicitPropertyOverridesSubclassDefault() { props.setProperty(DebeziumTransformerConfig.ENABLE_NESTED_FIELDS.key(), "false"); Dataset result = transformer.apply(jsc, spark, jsonToDataset(INSERT_EVENT, UPDATE_EVENT, DELETE_EVENT), props); - assertFalse(Arrays.asList(result.columns()).contains(DEBEZIUM_METADATA_FIELD)); + assertFalse(Arrays.asList(result.columns()).contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD)); } @Test @@ -215,6 +214,40 @@ void testSchemaAsNullableFalsePreservesNonNullableSourceFields() { assertEquals("Alice", result.first().getAs("name")); } + @Test + void testSchemaAsNullableFalseMakesMetadataColumnsNullableEvenIfSourceDeclaredNonNullable() { + // "op" is declared non-nullable at the envelope level, but it is a Debezium metadata column + // (not a __data-derived business column), so it must NOT be treated as a "known non-nullable + // source column" -- it should end up nullable in the output regardless of its declared + // nullability in the raw envelope schema. + StructType rowDataSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("id", DataTypes.LongType, false, Metadata.empty()), + DataTypes.createStructField("name", DataTypes.StringType, true, Metadata.empty()))); + StructType sourceSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("name", DataTypes.StringType, true, Metadata.empty()), + DataTypes.createStructField("ts_ms", DataTypes.LongType, true, Metadata.empty()))); + StructType envelopeSchema = DataTypes.createStructType(Arrays.asList( + DataTypes.createStructField("op", DataTypes.StringType, false, Metadata.empty()), + DataTypes.createStructField("ts_ms", DataTypes.LongType, true, Metadata.empty()), + DataTypes.createStructField("before", rowDataSchema, true, Metadata.empty()), + DataTypes.createStructField("after", rowDataSchema, true, Metadata.empty()), + DataTypes.createStructField("source", sourceSchema, true, Metadata.empty()))); + + Row afterRow = RowFactory.create(1L, "Alice"); + Row sourceRow = RowFactory.create("pgdb", 1700000000000L); + Row envelopeRow = RowFactory.create("c", 1700000000500L, null, afterRow, sourceRow); + Dataset input = spark.createDataFrame(Collections.singletonList(envelopeRow), envelopeSchema); + + TypedProperties props = new TypedProperties(); + props.setProperty(DebeziumTransformerConfig.SCHEMA_AS_NULLABLE.key(), "false"); + + Dataset result = new TestableDebeziumTransformer().apply(jsc, spark, input, props); + + assertTrue(result.schema().apply(DebeziumConstants.FLATTENED_OP_COL_NAME).nullable(), + "op is a Debezium metadata column, not a known non-nullable source column, so it must be nullable"); + assertFalse(result.schema().apply("id").nullable(), "id remains non-nullable as a known source column"); + } + @Test void testErrorTableEnabledAddsNullCorruptRecordWhenMissing() { // Include UPDATE_EVENT alongside so Spark's JSON inference sees a non-null `before` diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java index 93f3a4134a7cc..699a9aec7a7af 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestMysqlDebeziumTransformer.java @@ -78,7 +78,7 @@ void testFlatByDefault() { .apply(jsc, spark, jsonToDataset(mysqlEvent("c", 1, "mysql-bin.000001", 100)), new TypedProperties()); List columns = Arrays.asList(result.columns()); - assertFalse(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), + assertFalse(columns.contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD), "MySQL should be flat by default"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME), "_event_bin_file at root"); } @@ -92,12 +92,12 @@ void testNestedModeGroupsMetadataButSeqStaysCorrect() { .apply(jsc, spark, jsonToDataset(mysqlEvent("c", 1, "mysql-bin.000001", 100)), props); List columns = Arrays.asList(result.columns()); - assertTrue(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), "metadata nested"); + assertTrue(columns.contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD), "metadata nested"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op at root"); assertTrue(columns.contains(DebeziumConstants.ADDED_SEQ_COL_NAME), "_event_seq at root"); assertFalse(columns.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME), "binlog file is nested, not root"); - Row metadata = result.first().getAs(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD); + Row metadata = result.first().getAs(DebeziumConstants.DEBEZIUM_METADATA_FIELD); List nested = Arrays.asList(metadata.schema().fieldNames()); assertTrue(nested.contains(DebeziumConstants.FLATTENED_FILE_COL_NAME)); assertTrue(nested.contains(DebeziumConstants.FLATTENED_POS_COL_NAME)); diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java index 9d7fd680b38c9..9ad41a2d79b1e 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/debezium/TestPostgresDebeziumTransformer.java @@ -80,12 +80,12 @@ void testPostgresDefaultsToNestedMetadata() { .apply(jsc, spark, jsonToDataset(pgEvent("c", 1, 1001, false)), new TypedProperties()); List columns = Arrays.asList(result.columns()); - assertTrue(columns.contains(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD), "nested by default for Postgres"); + assertTrue(columns.contains(DebeziumConstants.DEBEZIUM_METADATA_FIELD), "nested by default for Postgres"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_OP_COL_NAME), "op at root"); assertTrue(columns.contains(DebeziumConstants.FLATTENED_LSN_COL_NAME), "LSN at root (payload ordering field)"); assertFalse(columns.contains(DebeziumConstants.FLATTENED_TX_ID_COL_NAME), "tx id is nested, not at root"); - Row metadata = result.first().getAs(AbstractDebeziumTransformer.DEBEZIUM_METADATA_FIELD); + Row metadata = result.first().getAs(DebeziumConstants.DEBEZIUM_METADATA_FIELD); assertTrue(Arrays.asList(metadata.schema().fieldNames()).contains(DebeziumConstants.FLATTENED_TX_ID_COL_NAME)); }