nsivabalan commented on code in PR #19110: URL: https://github.com/apache/hudi/pull/19110#discussion_r3669883973
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/MysqlDebeziumTransformer.java: ########## @@ -0,0 +1,77 @@ +/* + * 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. + * + * <p>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 "<binlog-file-suffix>.<pos>"} (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}. + * + * <p>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 { + + // Nestable MySQL metadata (grouped under _debezium_metadata when nesting is enabled). + private static final List<Column> MYSQL_METADATA = Arrays.asList( + new Column(DebeziumConstants.INCOMING_SOURCE_ROW_FIELD).alias(DebeziumConstants.FLATTENED_ROW_COL_NAME)); + + // The binlog coordinates are the payload's ordering fields, so they are kept at the root level in + // every layout (flat or nested), matching how the Postgres transformer keeps the LSN at the root. + private static final List<Column> MYSQL_ORDERING_COLUMNS = 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)); + + public MysqlDebeziumTransformer() { + super(MYSQL_METADATA, MYSQL_ORDERING_COLUMNS, 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"}. The binlog file and position are kept at the root + * level in both the flat and nested layouts, so they are read directly. + * + * @param dataset flattened MySQL Debezium dataset. + * @return dataset with the {@code _event_seq} column added. + */ + private static Dataset<Row> applySeqNo(Dataset<Row> dataset) { + return dataset.withColumn(DebeziumConstants.ADDED_SEQ_COL_NAME, functions.concat( + functions.substring_index(dataset.col(DebeziumConstants.FLATTENED_FILE_COL_NAME), ".", -1), + functions.lit("."), + dataset.col(DebeziumConstants.FLATTENED_POS_COL_NAME))); Review Comment: ⚠️ **IMPORTANT** — reopening this one. I don't think "pre-existing / out of scope" holds, and the consequence is worse than a degraded error message. `generateUniqueSequence` validated before building the key: ```java if (fileId == null || fileId.trim().isEmpty() || pos == null || pos < 0) { throw new HoodieReadFromSourceException( String.format("Invalid binlog file information from Debezium: fileId=%s, pos=%s", fileId, pos)); } ``` All three checks are gone here, and Spark `concat` returns **null** when any argument is null — so a malformed event silently yields a null `_event_seq`, the ordering field. Where that lands depends on which record it is in `MySqlDebeziumAvroPayload.shouldPickCurrentRecord`: - null on the **incoming** record → `orElseThrow` fires `HoodieDebeziumAvroPayloadException`. Still fails, but at merge time with an Avro record dumped into the message rather than `fileId=null, pos=…` at the source. - null on the **stored** record → `currentSourceSeqOpt` is empty, so it hits the `// handle bootstrap case` branch and `return false` — pick the incoming record unconditionally. A null seq is indistinguishable from a legitimately-bootstrapped row, so **ordering is silently skipped and an out-of-order event can overwrite a newer one.** That second path is why I'd like this addressed in the PR. It isn't just a worse message; under one arrangement it's a silent hole in the ordering guarantee that the precombine field exists to provide. `pos < 0` disappears too: a negative pos yields a well-formed-looking `"000001.-5"`, which `isCurrentSeqLatest` then compares as a string and orders wrong instead of failing. On scope — this is new code whose stated purpose is to replace the source-path flattening, so relative to the path it supersedes the guard was removed. That's a behavior regression introduced by this PR, not inherited from it. @wombatu-kun reached the same conclusion independently. A `when(col.isNull(), raise_error(...))` on the two coordinates, or an explicit validation before the concat, restores the fail-fast. Test: a MySQL envelope with `source.file` null (and one with `source.pos` null) should fail at transform time with a message naming the binlog coordinates. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/AbstractDebeziumTransformer.java: ########## @@ -0,0 +1,294 @@ +/* + * 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. + * + * <p>A Debezium change event is a nested record of the form + * {@code {op, ts_ms, before:{...}, after:{...}, source:{...}}}. This transformer: + * <ul> + * <li>selects the {@code before} image for deletes and the {@code after} image otherwise, + * and explodes it to the row's top level;</li> + * <li>surfaces the common Debezium metadata columns (operation type, processing/origin + * timestamps, shard) along with any database-specific metadata columns supplied by the + * subclass;</li> + * <li>optionally nests the metadata columns under a single {@code _debezium_metadata} struct + * (see {@link DebeziumTransformerConfig#ENABLE_NESTED_FIELDS});</li> + * <li>optionally preserves the error-table corrupt-record column when the error table is + * enabled;</li> + * <li>applies an optional database-specific post-processing step (e.g. ordering/sequence + * columns, LSN defaulting);</li> + * <li>normalizes column nullability (see + * {@link DebeziumTransformerConfig#SCHEMA_AS_NULLABLE}).</li> + * </ul> + * + * <p>The flattened column names are defined in {@link DebeziumConstants}; the matching + * {@code DebeziumAvroPayload} implementations rely on these names for merge/ordering semantics. + * + * <p>The layout and nullability behavior are configured through {@link DebeziumTransformerConfig}. + * + * <p>Subclasses configure the database-specific behavior purely through the constructor; there is + * no abstract method to implement. + */ +public class AbstractDebeziumTransformer implements Transformer { + + private static final String DATA_FIELD = "__data"; + // Bare name of the optional {@code schema} field inside the Debezium {@code source} struct + // (INCOMING_SOURCE_SCHEMA_FIELD is the fully-qualified {@code source.schema} path). + private static final String SOURCE_SCHEMA_FIELD_NAME = "schema"; + + private static final List<Column> DEFAULT_ROOT_LEVEL_METADATA_COLUMNS = Arrays.asList( + new Column(DebeziumConstants.INCOMING_OP_FIELD).alias(DebeziumConstants.FLATTENED_OP_COL_NAME)); + + private static final List<Column> 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<Column> typeSpecificMetadataColumns; + private final List<Column> rootLevelOrderingColumns; + private final Option<Function<Dataset<Row>, Dataset<Row>>> postProcessingOption; + private final boolean nestedFieldsEnabledByDefault; + + protected AbstractDebeziumTransformer( + List<Column> typeSpecificMetadataColumns, + List<Column> rootLevelOrderingColumns, + Option<Function<Dataset<Row>, Dataset<Row>>> postProcessingOption) { + this(typeSpecificMetadataColumns, rootLevelOrderingColumns, postProcessingOption, false); + } + + /** + * @param typeSpecificMetadataColumns database-specific metadata columns (already aliased to their + * flattened output names) that are grouped under the + * {@code _debezium_metadata} struct when nesting is enabled. + * @param rootLevelOrderingColumns database-specific ordering / log-position columns (e.g. the + * Postgres LSN, or the MySQL binlog file and position) that stay + * at the root level in every layout, so the payload's ordering + * field is always a root-level column and needs no nested-path + * handling. + * @param postProcessingOption optional post-flatten transformation applied to the result. + * @param nestedFieldsEnabledByDefault the subclass default for the metadata layout. Resolution + * order at runtime: an explicitly set + * {@code hoodie.streamer.transformer.debezium.nested.fields.enable} + * ({@link DebeziumTransformerConfig#ENABLE_NESTED_FIELDS}) + * always wins; when the property is absent this default is + * used. Lets a subclass opt into nested metadata by default. + */ + protected AbstractDebeziumTransformer( + List<Column> typeSpecificMetadataColumns, + List<Column> rootLevelOrderingColumns, + Option<Function<Dataset<Row>, Dataset<Row>>> postProcessingOption, + boolean nestedFieldsEnabledByDefault) { + this.typeSpecificMetadataColumns = typeSpecificMetadataColumns; + this.rootLevelOrderingColumns = rootLevelOrderingColumns; + this.postProcessingOption = postProcessingOption; + this.nestedFieldsEnabledByDefault = nestedFieldsEnabledByDefault; + } + + @Override + public Dataset<Row> apply(JavaSparkContext javaSparkContext, SparkSession sparkSession, Dataset<Row> rowDataset, TypedProperties props) { + if (rowDataset.columns().length == 0) { + return rowDataset; + } + Dataset<Row> withDataField = selectBeforeOrAfterImage(rowDataset); + List<Column> outputColumns = buildOutputColumns(withDataField, props); + Dataset<Row> withErrorCol = applyErrorTablePassthrough(withDataField, outputColumns, props); + Dataset<Row> flattened = withErrorCol.select(outputColumns.toArray(new Column[]{})); + Dataset<Row> postProcessed = postProcessingOption.map(postProcessing -> postProcessing.apply(flattened)).orElse(flattened); + return applyNullabilityRules(sparkSession, withDataField, postProcessed, props); + } + + /** + * Selects the {@code before} image for deletes and the {@code after} image otherwise into a single + * {@code __data} struct column, then drops the original {@code before}/{@code after} columns. + */ + private static Dataset<Row> selectBeforeOrAfterImage(Dataset<Row> rowDataset) { + return 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); + } + + /** + * Builds the flattened output column list: the metadata columns (flat at the root or grouped under + * the {@code _debezium_metadata} struct, per {@link DebeziumTransformerConfig#ENABLE_NESTED_FIELDS}) + * followed by the exploded {@code __data} image. + */ + private List<Column> buildOutputColumns(Dataset<Row> withDataField, TypedProperties props) { + List<Column> outputColumns = new ArrayList<>(); + if (isNestedFieldsEnabled(props)) { + outputColumns.addAll(buildNestedMetadataColumns(withDataField)); + } else { + // When nested fields are disabled, all metadata fields are at the root level. + outputColumns.addAll(DEFAULT_ROOT_LEVEL_METADATA_COLUMNS); + outputColumns.addAll(DEFAULT_NESTED_METADATA_COLUMNS); + outputColumns.addAll(typeSpecificMetadataColumns); + outputColumns.addAll(rootLevelOrderingColumns); + } + // Explode the selected before/after image to the row's top level. + outputColumns.add(new Column(String.format("%s.*", DATA_FIELD))); + return outputColumns; + } + + /** + * Assembles the metadata columns for the nested layout: the operation-type column and the + * database-specific ordering / log-position columns (e.g. the Postgres LSN, or the MySQL binlog + * file and position) stay at the root level so payload ordering keeps working against root-level + * columns, while every other metadata column is grouped under the {@code _debezium_metadata} struct. + */ + private List<Column> buildNestedMetadataColumns(Dataset<Row> withDataField) { + List<Column> nestedMetadataFields = new ArrayList<>(DEFAULT_NESTED_METADATA_COLUMNS); + nestedMetadataFields.addAll(typeSpecificMetadataColumns); + // Only add the schema field if it exists in the source struct (not all databases have this field). + if (hasSchemaField(withDataField)) { + nestedMetadataFields.add(new Column(DebeziumConstants.INCOMING_SOURCE_SCHEMA_FIELD).alias(DebeziumConstants.FLATTENED_SCHEMA_NAME)); + } + + List<Column> outputColumns = new ArrayList<>(); + outputColumns.add(functions.struct(nestedMetadataFields.toArray(new Column[]{})) + .alias(DebeziumConstants.DEBEZIUM_METADATA_FIELD)); + // The operation-type column and the ordering / log-position columns stay at the root level. + outputColumns.addAll(DEFAULT_ROOT_LEVEL_METADATA_COLUMNS); + outputColumns.addAll(rootLevelOrderingColumns); + return outputColumns; + } + + /** + * When the error table is enabled, ensures the corrupt-record column is present (adding a null one + * if the input lacks it) and includes it in {@code outputColumns} so it is preserved downstream. + */ + private static Dataset<Row> applyErrorTablePassthrough(Dataset<Row> dataset, List<Column> outputColumns, TypedProperties props) { + if (!ConfigUtils.getBooleanWithAltKeys(props, ERROR_TABLE_ENABLED)) { + return dataset; + } + Dataset<Row> withCorruptCol = dataset; + if (!Arrays.asList(dataset.columns()).contains(ERROR_TABLE_CURRUPT_RECORD_COL_NAME)) { + withCorruptCol = dataset.withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME, functions.lit(null)); + } + outputColumns.add(new Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME)); + return withCorruptCol; + } + + /** + * Normalizes column nullability on the flattened dataset. When + * {@link DebeziumTransformerConfig#SCHEMA_AS_NULLABLE} is set every column is marked nullable; + * otherwise a column stays non-nullable only if Spark already infers it non-nullable or if it was a + * non-nullable source data column. This preserves the non-nullability of Debezium metadata columns + * (e.g. {@code _change_operation_type}) that Spark infers as non-nullable. + * + * @param withDataField the dataset carrying the {@code __data} struct, used to recover which source + * data columns were non-nullable before flattening. + */ + private Dataset<Row> applyNullabilityRules(SparkSession sparkSession, Dataset<Row> withDataField, + Dataset<Row> debeziumDataset, TypedProperties props) { + if (ConfigUtils.getBooleanWithAltKeys(props, DebeziumTransformerConfig.SCHEMA_AS_NULLABLE)) { + return convertColumnsToNullable(sparkSession, debeziumDataset); + } + + Set<String> nonNullableColumns = new HashSet<>(); + for (StructField field : withDataField.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())); + } + } + + 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())) Review Comment: ⚠️ **IMPORTANT** — this ternary can stamp `nullable=false` onto a column that genuinely holds nulls, producing a schema that lies about the data. Read it as a truth table, where `field` is a **root-level** column of the already-flattened dataset: | `field.nullable()` (root) | in `nonNullableColumns`? | branch | result | |---|---|---|---| | true | no | then | `nullable=true` ✅ | | false | no | else | `nullable=false` ✅ | | false | yes | else | `nullable=false` ✅ | | **true** | **yes** | **else** | **`nullable=false`** ❌ | The last row is the problem. `nonNullableColumns` is collected from the fields *inside* the `__data` struct, but non-nullable-inside-the-struct does not imply non-null-at-root. Flattening is `__data.*` over ```java when(op == "d", col("before")).otherwise(col("after")) ``` and `struct.*` expansion over a **null struct** yields null for every expanded field — the inner declared nullability doesn't protect you, because the struct itself is null. Debezium emits `before=null` for deletes whenever Postgres `REPLICA IDENTITY` isn't `FULL`, and `DEFAULT` is the Postgres default. So on an ordinary configuration, every delete row has nulls in columns this method just marked non-nullable. Nothing catches it here: `createDataFrame(RDD<Row>, StructType)` does not validate rows against the schema, it just attaches it. The lie surfaces later as a null-in-required-field error from the Avro/Parquet writer, pointing nowhere near this method. Worth noting the code this replaces only ever **widened** — `DebeziumSource.convertColumnToNullable` passes non-matching fields through untouched (`: field`) and never narrows. The narrowing is new here. This only bites when `schema.nullable.enable=false`, and the default is `true`, so the default path takes the early `convertColumnsToNullable` return and is safe. That's why this is IMPORTANT rather than blocking — but a non-default config silently emitting a wrong schema is the kind of thing that gets diagnosed months later, so I'd rather fix it than defer. Simplest correct change is to stop narrowing — preserve whatever Spark inferred: ```java .map(field -> field.nullable() && !nonNullableColumns.contains(field.name()) ? new StructField(field.name(), field.dataType(), true, field.metadata()) : field) ``` That raises a design question though: the javadoc motivates this branch with `_change_operation_type`, but Spark already infers that non-nullable when `op` is non-nullable. If inference handles the cited case on its own, is the whole `nonNullableColumns` computation earning its keep, or could this method reduce to "widen everything, or do nothing"? Up front: **this is from code inspection — I wasn't able to confirm it at runtime** (the `spark3.5` profile wouldn't compile `hudi-utilities` against my local artifacts, unrelated to this PR). The load-bearing step is the `struct.*`-over-null-struct behavior; please verify that with a test rather than taking my word for it. Suggested test — the existing two nullability tests both use a single insert with a non-null `after`, so neither reaches this row of the table: delete envelope with `before=null`, a source column declared `nullable=false` inside the struct, `schema.nullable.enable=false`. Assert the root column is still nullable, and round-trip through an actual `write().save()` so the writer validates the schema against the data. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/debezium/PostgresDebeziumTransformer.java: ########## @@ -0,0 +1,84 @@ +/* + * 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. + * + * <p>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}. + * + * <p>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 <a href="https://debezium.io/blog/2021/10/07/incremental-snapshots/";>incremental snapshots</a>). + * + * <p>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. When nested, the {@code _event_lsn} and operation-type + * columns stay at the root level so payload ordering keeps working. + */ +public class PostgresDebeziumTransformer extends AbstractDebeziumTransformer { + + // Nestable Postgres metadata (grouped under _debezium_metadata when nesting is enabled). + private static final List<Column> POSTGRES_METADATA = Arrays.asList( + new Column(DebeziumConstants.INCOMING_SOURCE_TXID_FIELD).alias(DebeziumConstants.FLATTENED_TX_ID_COL_NAME), + new Column(DebeziumConstants.INCOMING_SOURCE_XMIN_FIELD).alias(DebeziumConstants.FLATTENED_XMIN_COL_NAME)); + + // The LSN is the payload's ordering field, so it is kept at the root level in every layout. + private static final List<Column> POSTGRES_ORDERING_COLUMNS = Arrays.asList( + new Column(DebeziumConstants.INCOMING_SOURCE_LSN_FIELD).alias(DebeziumConstants.FLATTENED_LSN_COL_NAME)); + + public PostgresDebeziumTransformer() { + super(POSTGRES_METADATA, POSTGRES_ORDERING_COLUMNS, Option.of(PostgresDebeziumTransformer::useDefaultValuesForLsnIfNull)); Review Comment: 💬 Could you refresh the PR description before merge? It's drifted from the code across the last two commits, which makes the change hard to review from the summary alone: - It still describes `nestedDebeziumMetadataEnabled` being threaded through new `inferMergingConfigsForWrites` / `inferMergingConfigsForV9TableCreation` overloads, and `HoodieStreamer` supplying the value — all reverted in `2a4926609787`. The `HoodieTableConfig` diff is now just a comment typo fix. - It says `PostgresDebeziumTransformer` "Nests metadata by default" and `MysqlDebeziumTransformer` is "Flat by default," but both now call the 3-arg `super(...)`, so `nestedFieldsEnabledByDefault=false` for both. The `ENABLE_NESTED_FIELDS` documentation already says "both the Postgres and MySQL transformers default to flat metadata" — the description is the stale one. One consequence worth a second look while you're in there: with both transformers now defaulting to flat, and the `source.schema` → `db_schema_source_partition` column only emitted in the nested branch, the **default** path no longer surfaces it — whereas `PostgresDebeziumSource.processDataset` always emitted it at root. It's not in `META_COLUMNS` and has no in-repo consumer, so blast radius is limited to external pipelines keyed on it, but the parity gap moved from an opt-in path to the default one when the nesting default flipped. Honoring `hasSchemaField` in the flat branch too would close it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
