This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 87b86fef303de044ac3d74e1663c69243dc7f9e5 Author: Aditya Goenka <[email protected]> AuthorDate: Sat Jun 13 09:54:01 2026 +0530 fix(spark): strip _hoodie_* meta columns from CDC before/after images (#18948) * [HUDI-14363] Strip _hoodie_* meta columns from CDC before/after images CDC before/after images inferred directly from base/log data files (e.g. the BASE_FILE_INSERT case hit by an insert-only commit that writes no CDC log file) leaked the _hoodie_* meta columns, while images served from the supplemental CDC log already have them stripped at write time. This produced an inconsistent, alternating-per-commit image schema. --------- Co-authored-by: Claude Opus 4.8 <[email protected]> (cherry picked from commit 97c03f75c737ac88d8ac7f5234a9cf3856d05cfb) --- .../org/apache/hudi/cdc/CDCFileGroupIterator.scala | 22 ++++++-- .../functional/cdc/TestCDCDataFrameSuite.scala | 61 ++++++++++++++++++++++ 2 files changed, 79 insertions(+), 4 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala index 4970be3c249b..cc316243d266 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala @@ -62,6 +62,7 @@ import org.apache.spark.unsafe.types.UTF8String import java.io.Closeable import java.util import java.util.{Collections, Locale} +import java.util.function.UnaryOperator import java.util.stream.Collectors import scala.annotation.tailrec @@ -228,7 +229,9 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit, props.getBoolean(DISK_MAP_BITCASK_COMPRESSION_ENABLED.key(), DISK_MAP_BITCASK_COMPRESSION_ENABLED.defaultValue()), getClass.getSimpleName) - private val internalRowToJsonStringConverterMap: mutable.Map[Integer, InternalRowToJsonStringConverter] = mutable.Map.empty + // Per-schema cache of the (image projection, json converter) used to build CDC before/after + // images. Keyed by the record's schema id so schema evolution is handled correctly. + private val cdcImageConverterMap: mutable.Map[Integer, (UnaryOperator[InternalRow], InternalRowToJsonStringConverter)] = mutable.Map.empty private def needLoadNextFile: Boolean = { !recordIter.hasNext && @@ -559,9 +562,20 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit, * Convert InternalRow to json string. */ private def convertBufferedRecordToJsonString(record: BufferedRecord[InternalRow]): UTF8String = { - internalRowToJsonStringConverterMap.getOrElseUpdate(record.getSchemaId, - new InternalRowToJsonStringConverter(HoodieInternalRowUtils.getCachedSchema(readerContext.getRecordContext.decodeAvroSchema(record.getSchemaId)))) - .convert(record.getRecord) + val (imageProjection, converter) = cdcImageConverterMap.getOrElseUpdate(record.getSchemaId, { + val recordSchema = readerContext.getRecordContext.decodeAvroSchema(record.getSchemaId) + // CDC before/after images must contain only business columns. Records read from base/log + // files carry the _hoodie_* meta columns (kept on the InternalRow because they are needed + // internally for record keying and merging), while images served from the supplemental CDC + // log already have them stripped at write time (HoodieCDCLogger). Project each record onto + // the meta-stripped image schema so every inference case produces a schema-consistent, + // business-columns-only image. + val imageSchema = HoodieSchemaUtils.removeMetadataFields(recordSchema) + val projection = readerContext.getRecordContext.projectRecord(recordSchema, imageSchema) + val converter = new InternalRowToJsonStringConverter(HoodieInternalRowUtils.getCachedSchema(imageSchema)) + (projection, converter) + }) + converter.convert(imageProjection.apply(record.getRecord)) } /** diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala index 51905c00caa4..4bada04ab29d 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala +++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/TestCDCDataFrameSuite.scala @@ -21,6 +21,7 @@ package org.apache.hudi.functional.cdc import org.apache.hudi.DataSourceWriteOptions import org.apache.hudi.DataSourceWriteOptions.{MOR_TABLE_TYPE_OPT_VAL, PARTITIONPATH_FIELD_OPT_KEY, PRECOMBINE_FIELD_OPT_KEY, RECORDKEY_FIELD_OPT_KEY} import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs +import org.apache.hudi.common.model.HoodieRecord import org.apache.hudi.common.table.{HoodieTableConfig, TableSchemaResolver} import org.apache.hudi.common.table.cdc.{HoodieCDCOperation, HoodieCDCSupplementalLoggingMode} import org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode.OP_KEY_ONLY @@ -959,4 +960,64 @@ class TestCDCDataFrameSuite extends HoodieCDCTestBase { } assertTrue(newRecordFound, "Should have found the new record with complex data types in CDC") } + + /** + * Regression test for HUDI-14363: the CDC incremental query must produce before/after images + * that contain only business columns, never the _hoodie_* meta columns. Before the fix, images + * inferred directly from base/log data files (e.g. the BASE_FILE_INSERT case, which is hit by an + * insert-only commit that writes no CDC log file) leaked the meta columns, while images served + * from the supplemental CDC log did not - producing an inconsistent, alternating-per-commit + * schema. This reproduces the reporter's scenario (MOR + inline compaction every delta commit + + * upsert inserts) and asserts no image carries meta fields for any supplemental logging mode. + */ + @ParameterizedTest + @EnumSource(classOf[HoodieCDCSupplementalLoggingMode]) + def testCDCImagesExcludeHoodieMetaFields(loggingMode: HoodieCDCSupplementalLoggingMode): Unit = { + val options = commonOpts ++ Map( + DataSourceWriteOptions.TABLE_TYPE.key() -> DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL, + HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE.key -> loggingMode.name(), + "hoodie.compact.inline" -> "true", + "hoodie.compact.inline.max.delta.commits" -> "1" + ) + + // 1. Insert - this commit writes no CDC log file, so its change data is inferred from the base + // file via the BASE_FILE_INSERT case (the path that previously leaked _hoodie_* meta columns). + val records1 = recordsToStrings(dataGen.generateInserts("000", 100)).asScala.toList + spark.read.json(spark.sparkContext.parallelize(records1, 2)) + .write.format("org.apache.hudi") + .options(options) + .mode(SaveMode.Overwrite) + .save(basePath) + metaClient = createMetaClient(spark, basePath) + val instant1 = metaClient.reloadActiveTimeline.lastInstant().get() + assertFalse(hasCDCLogFile(instant1)) + val commitTime1 = instant1.requestedTime + + // 2. Upsert (updates + new inserts) - exercises the supplemental CDC log (AS_IS) path too. + val updates = recordsToStrings(dataGen.generateUniqueUpdates("001", 30)).asScala.toList + val inserts = recordsToStrings(dataGen.generateInserts("001", 20)).asScala.toList + spark.read.json(spark.sparkContext.parallelize(updates ++ inserts, 2)) + .write.format("org.apache.hudi") + .options(options) + .mode(SaveMode.Append) + .save(basePath) + + // Read all change data and assert no before/after image contains a Hudi meta column. Note we + // check the actual meta-column names rather than the "_hoodie_" prefix: _hoodie_is_deleted is a + // business/payload field (the soft-delete marker carried in the record schema), not a meta + // column, so it is expected to remain in the image. + val allCDCData = cdcDataFrame((commitTime1.toLong - 1).toString).collect() + assertTrue(allCDCData.nonEmpty, "Expected some CDC rows") + allCDCData.foreach { row => + Seq("before", "after").foreach { col => + val json = row.getAs[String](col) + if (json != null) { + HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.asScala.foreach { metaCol => + assertFalse(json.contains(metaCol), + s"$col image should not contain meta column $metaCol, but was: $json") + } + } + } + } + } }
