You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/05 11:10:03 UTC
(hudi) 02/05: fix(spark): strip _hoodie_* meta columns from CDC before/after images (#18948)
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 <63...@users.noreply.github.com>
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 <no...@anthropic.com>
(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")
+ }
+ }
+ }
+ }
+ }
}