This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 6bbf798f88d4 fix(spark): render decimal, date and timestamp values in
CDC images (#20305)
6bbf798f88d4 is described below
commit 6bbf798f88d468d1ca3ce774aae991145b0cd0a2
Author: Sagar Sumit <[email protected]>
AuthorDate: Wed Oct 14 17:02:44 2026 +0530
fix(spark): render decimal, date and timestamp values in CDC images (#20305)
CDC before/after images rendered DECIMAL as {"zero":false} (Spark's
Decimal serialized as a bean) or, for inline Avro CDC log files, as the
raw unscaled bytes. DATE and TIMESTAMP came out as epoch days/micros.
InternalRowToJsonStringConverter now renders DECIMAL as an exact JSON
number, DATE as an ISO-8601 date, TIMESTAMP as an ISO-8601 UTC instant
and TIMESTAMP_NTZ as an ISO-8601 local date-time, at full precision.
Images from inline Avro CDC log files are deserialized to InternalRow
and rendered by the same converter, so every CDC inference path
produces the same image. HoodieCDCUtils.recordToJson has no remaining
callers and is removed.
Fixes #20299
---
.../hudi/common/table/cdc/HoodieCDCUtils.java | 4 --
.../org/apache/hudi/cdc/CDCFileGroupIterator.scala | 10 ++--
.../cdc/InternalRowToJsonStringConverter.scala | 17 +++++--
.../cdc/TestInternalRowToJsonStringConverter.scala | 51 +++++++++++++++++++-
.../functional/cdc/TestCDCDataFrameSuite.scala | 56 +++++++++++++++++++++-
5 files changed, 126 insertions(+), 12 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java
index 9792b5adfea0..feb9eb2a933f 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java
@@ -155,8 +155,4 @@ public class HoodieCDCUtils {
record.put(CDC_RECORD_KEY, recordKey);
return record;
}
-
- public static String recordToJson(GenericRecord record) {
- return GenericData.get().toString(record);
- }
}
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 eba103726ef3..9e0cc1a6fadb 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
@@ -248,7 +248,10 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
org.apache.hudi.common.util.Option.of(parquetSchema)
}
- private lazy val nativeCdcImageConverter = new
InternalRowToJsonStringConverter(cdcDataSparkSchema)
+ private lazy val cdcImageConverter = new
InternalRowToJsonStringConverter(cdcDataSparkSchema)
+
+ private lazy val inlineCdcImageDeserializer =
sparkAdapter.createAvroDeserializer(
+ HoodieSchemaUtils.removeMetadataFields(schema), cdcDataSparkSchema)
private def needLoadNextFile: Boolean = {
!recordIter.hasNext &&
@@ -627,7 +630,7 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
private def cdcRecordImageToJson(record: HoodieCDCLogRecord[_], ordinal:
Int): UTF8String = {
if (record.isNative) {
val image = record.getEngineImage(ordinal,
cdcDataSparkSchema.length).asInstanceOf[InternalRow]
- if (image == null) null else nativeCdcImageConverter.convert(image)
+ if (image == null) null else cdcImageConverter.convert(image)
} else {
recordToJsonAsUTF8String(record.getAvroImage(ordinal).asInstanceOf[GenericRecord])
}
@@ -664,7 +667,8 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
if (record == null) {
null
} else {
- convertToUTF8String(HoodieCDCUtils.recordToJson(record))
+ val row =
inlineCdcImageDeserializer.deserialize(record).get.asInstanceOf[InternalRow]
+ cdcImageConverter.convert(row)
}
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
index 4d01831e9a8c..c89ebbb33200 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala
@@ -19,15 +19,18 @@
package org.apache.hudi.cdc
import com.fasterxml.jackson.annotation.JsonInclude.Include
-import com.fasterxml.jackson.core.JsonProcessingException
+import com.fasterxml.jackson.core.{JsonGenerator, JsonProcessingException}
import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import com.fasterxml.jackson.databind.util.RawValue
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType,
StructType}
+import org.apache.spark.sql.catalyst.util.{ArrayData, DateTimeUtils, MapData}
+import org.apache.spark.sql.types.{ArrayType, DataType, DateType, Decimal,
DecimalType, MapType, StringType, StructType, TimestampType}
import org.apache.spark.unsafe.types.UTF8String
+import java.time.LocalDate
+import java.time.format.DateTimeFormatter
+
class InternalRowToJsonStringConverter(schema: StructType) {
private lazy val mapper: ObjectMapper = {
@@ -37,6 +40,7 @@ class InternalRowToJsonStringConverter(schema: StructType) {
// The variant branch below embeds its input verbatim once readTree
accepts it, and readTree on
// its own is happy to parse a valid prefix and leave the rest unconsumed.
_mapper.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true)
+ _mapper.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, true)
_mapper.registerModule(DefaultScalaModule)
_mapper
}
@@ -95,6 +99,11 @@ class InternalRowToJsonStringConverter(schema: StructType) {
structMap.toMap
case _ => value // fallback
}
+ case _: DecimalType => value.asInstanceOf[Decimal].toJavaBigDecimal
+ case DateType => LocalDate.ofEpochDay(value.asInstanceOf[Int]).toString
+ case TimestampType =>
DateTimeUtils.microsToInstant(value.asInstanceOf[Long]).toString
+ case dt if dt.typeName ==
InternalRowToJsonStringConverter.TIMESTAMP_NTZ_TYPE_NAME =>
+
DateTimeUtils.microsToLocalDateTime(value.asInstanceOf[Long]).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
case dt if dt.typeName ==
InternalRowToJsonStringConverter.VARIANT_TYPE_NAME =>
// VariantVal.toString renders the variant as JSON; embed that
rendering verbatim so the
// image carries the variant's structure. Falling through to the
default would serialize
@@ -141,4 +150,6 @@ object InternalRowToJsonStringConverter {
* against Spark 3 where the type does not exist, needs neither the symbol
nor a SparkAdapter.
*/
private val VARIANT_TYPE_NAME = "variant"
+
+ private val TIMESTAMP_NTZ_TYPE_NAME = "timestamp_ntz"
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
index 7ea8cae4bc40..1a5daf8ed928 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/cdc/TestInternalRowToJsonStringConverter.scala
@@ -24,7 +24,7 @@ import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData}
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, MapType,
Metadata, StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, Decimal,
MapType, Metadata, StructField, StructType}
import org.apache.spark.unsafe.types.UTF8String
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api.Assumptions.assumeTrue
@@ -116,6 +116,32 @@ class TestInternalRowToJsonStringConverter {
assertEquals("""{"id":4,"name":"test_nested","people":[{"name":"Alice","age":30},{"name":"Bob","age":25}]}""",
converted.toString)
}
+ @Test
+ def logicalTypes(): Unit = {
+ val row = InternalRow.fromSeq(Seq(
+ 1,
+ Decimal(BigDecimal("9.99"), 7, 2),
+ Decimal(BigDecimal("0.00000012"), 10, 8),
+ 20739,
+ 1791910800000000L,
+ 1791910800123456L,
+ ArrayData.toArrayData(Array(Decimal(BigDecimal("1.20"), 7, 2),
Decimal(BigDecimal("3.00"), 7, 2)))
+ ))
+ val converted = new
InternalRowToJsonStringConverter(logicalTypesSchema.structTypeSchema).convert(row)
+
assertEquals("""{"id":1,"amount":9.99,"rate":0.00000012,"day":"2026-10-13","ts":"2026-10-13T17:00:00Z","""
+ +
""""ts_micros":"2026-10-13T17:00:00.123456Z","amounts":[1.20,3.00]}""",
converted.toString)
+ }
+
+ @Test
+ def timestampNtzType(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark3_4, "TimestampNTZType requires Spark
3.4 or higher")
+ val schema = new StructType(Array[StructField](
+ StructField("id", DataTypes.IntegerType, nullable = false,
Metadata.empty),
+ StructField("ts", DataType.fromDDL("timestamp_ntz"), nullable = true,
Metadata.empty)))
+ val row = InternalRow.fromSeq(Seq(1, 1791885600000000L))
+ assertEquals("""{"id":1,"ts":"2026-10-13T10:00:00"}""", new
InternalRowToJsonStringConverter(schema).convert(row).toString)
+ }
+
@Test
def arrayWithNulls(): Unit = {
val row = InternalRow.fromSeq(Seq(
@@ -242,6 +268,29 @@ class TestInternalRowToJsonStringConverter {
HoodieTableSchema(structTypeSchema, HoodieSchema.parse(avroSchemaStr),
Option.empty[InternalSchema])
}
+ private def logicalTypesSchema: HoodieTableSchema = {
+ val structTypeSchema = new StructType(Array[StructField](
+ StructField("id", DataTypes.IntegerType, nullable = false,
Metadata.empty),
+ StructField("amount", DataTypes.createDecimalType(7, 2), nullable =
true, Metadata.empty),
+ StructField("rate", DataTypes.createDecimalType(10, 8), nullable = true,
Metadata.empty),
+ StructField("day", DataTypes.DateType, nullable = true, Metadata.empty),
+ StructField("ts", DataTypes.TimestampType, nullable = true,
Metadata.empty),
+ StructField("ts_micros", DataTypes.TimestampType, nullable = true,
Metadata.empty),
+ StructField("amounts", ArrayType(DataTypes.createDecimalType(7, 2)),
nullable = true, Metadata.empty)))
+ val avroSchemaStr: String =
+ """{"type": "record", "name": "test", "fields": [
+ |{"name": "id", "type": "int"},
+ |{"name": "amount", "type": {"type": "fixed", "name": "amount",
"size": 4, "logicalType": "decimal", "precision": 7, "scale": 2}},
+ |{"name": "rate", "type": {"type": "fixed", "name": "rate", "size": 5,
"logicalType": "decimal", "precision": 10, "scale": 8}},
+ |{"name": "day", "type": {"type": "int", "logicalType": "date"}},
+ |{"name": "ts", "type": {"type": "long", "logicalType":
"timestamp-micros"}},
+ |{"name": "ts_micros", "type": {"type": "long", "logicalType":
"timestamp-micros"}},
+ |{"name": "amounts", "type": {"type": "array", "items": {"type":
"fixed", "name": "amounts", "size": 4,
+ | "logicalType": "decimal", "precision": 7, "scale": 2}}}
+ |]}""".stripMargin
+ HoodieTableSchema(structTypeSchema, HoodieSchema.parse(avroSchemaStr),
Option.empty[InternalSchema])
+ }
+
private def hoodieTableSchema: HoodieTableSchema = {
val structTypeSchema = new StructType(Array[StructField](
StructField("uuid", DataTypes.IntegerType, nullable = false,
Metadata.empty),
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 853d82313740..2cc5ab35ed2d 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
@@ -33,12 +33,14 @@ import org.apache.hudi.exception.HoodieException
import org.apache.avro.generic.GenericRecord
import org.apache.spark.sql.{Row, SaveMode}
-import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType, MapType,
StringType, StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, DateType, DecimalType,
IntegerType, LongType, MapType, StringType, StructField, StructType,
TimestampType}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertThrows, assertTrue}
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.{CsvSource, EnumSource}
+import java.time.{Instant, LocalDate}
+
import scala.collection.JavaConverters._
class TestCDCDataFrameSuite extends HoodieCDCTestBase {
@@ -1051,4 +1053,56 @@ class TestCDCDataFrameSuite extends HoodieCDCTestBase {
cdcDataFrame(instantBefore(commitTime1)).collect().map(_.toString).sorted.toSeq,
cdcFromEarliest.collect().map(_.toString).sorted.toSeq)
}
+
+ @ParameterizedTest
+ @CsvSource(Array(
+ "COPY_ON_WRITE,data_before_after,TEN",
"MERGE_ON_READ,data_before_after,TEN",
+ "COPY_ON_WRITE,data_before,TEN", "MERGE_ON_READ,data_before,TEN",
+ "COPY_ON_WRITE,op_key_only,TEN", "MERGE_ON_READ,op_key_only,TEN",
+ "COPY_ON_WRITE,data_before_after,NINE",
"MERGE_ON_READ,data_before_after,NINE",
+ "COPY_ON_WRITE,data_before,NINE", "MERGE_ON_READ,data_before,NINE",
+ "COPY_ON_WRITE,op_key_only,NINE", "MERGE_ON_READ,op_key_only,NINE"))
+ def testCDCImagesWithLogicalTypes(tableType: String, loggingMode: String,
tableVersion: HoodieTableVersion): Unit = {
+ val schema = StructType(Seq(
+ StructField("_row_key", StringType, nullable = false),
+ StructField("timestamp", LongType, nullable = false),
+ StructField("amt", DecimalType(7, 2), nullable = true),
+ StructField("d", DateType, nullable = true),
+ StructField("ts", TimestampType, nullable = true)
+ ))
+ val options = commonOpts ++ Map(
+ HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE.key -> loggingMode,
+ DataSourceWriteOptions.TABLE_TYPE.key() -> tableType,
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
tableVersion.versionCode().toString,
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false"
+ )
+ val day = LocalDate.of(2026, 10, 13)
+ val ts = Instant.parse("2026-10-13T17:00:00.123456Z")
+
+ val inserts = (1 to 3).map(i => Row(i.toString, 1L, new
java.math.BigDecimal("1.25"), day, ts))
+ spark.createDataFrame(inserts.asJava, schema)
+ .write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+ metaClient = createMetaClient(spark, basePath)
+ val commitTime1 =
metaClient.reloadActiveTimeline.lastInstant().get().requestedTime
+
+ val update = Seq(Row("1", 2L, new java.math.BigDecimal("9.99"), day, ts))
+ spark.createDataFrame(update.asJava, schema)
+ .write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val cdcData = cdcDataFrame(instantBefore(commitTime1))
+ assertCDCOpCnt(cdcData, 3, 1, 0)
+ val updateRow = cdcData.where("op = 'u'").select("before", "after").head()
+ assertEquals(
+
"""{"_row_key":"1","timestamp":1,"amt":1.25,"d":"2026-10-13","ts":"2026-10-13T17:00:00.123456Z"}""",
+ updateRow.getString(0))
+ assertEquals(
+
"""{"_row_key":"1","timestamp":2,"amt":9.99,"d":"2026-10-13","ts":"2026-10-13T17:00:00.123456Z"}""",
+ updateRow.getString(1))
+ }
}
