codope opened a new issue, #20299:
URL: https://github.com/apache/hudi/issues/20299
### Bug Description
**What happened:**
In Spark CDC queries (`hudi_table_changes(..., 'cdc', ...)`, or the
DataFrame reader with
`hoodie.datasource.query.incremental.format=cdc`), DECIMAL values in the
`before` and `after`
JSON images are wrong. The symptom depends on where the image comes from:
- **Merge-on-Read** (images inferred from base and log files): every DECIMAL
is written as
`{"zero":false}` (or `{"zero":true}`). The value is lost.
- **Copy-on-Write** (images from the supplemental CDC log): every DECIMAL is
written as the raw
bytes of its unscaled value, for example `[0, 0, 3, -25]` for 9.99. You
can only recover the
value if you already know the scale.
DATE and TIMESTAMP values come out as days since epoch and epoch
microseconds on both table
types. They are readable, but don't match the column types.
Downstream consumers of CDC (SCD tables, audit trails, sinks that parse the
images) can't use
any decimal column. Money columns are usually decimals.
**What you expected:**
The images carry the column values in a form JSON readers understand, for
example
`"amt": 9.99` (or `"9.99"`), and the same representation on both table
types. Dates and
timestamps would ideally be ISO strings, or at least documented.
**Steps to reproduce:**
```scala
// spark-shell with hudi-spark3.5-bundle_2.12:1.2.1 and the usual Hudi
session extensions
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SaveMode
def run(tableType: String, path: String) = {
val df = spark.range(0, 3).select(col("id"),
lit(java.math.BigDecimal.valueOf(1.25)).cast("decimal(7,2)").as("amt"),
lit("2026-10-13").cast("date").as("d"),
lit("2026-10-13 10:00:00").cast("timestamp").as("ts"),
lit(1).as("qty"))
val o = Map(
"hoodie.table.name" -> s"cdc_types_$tableType",
"hoodie.datasource.write.recordkey.field" -> "id",
"hoodie.datasource.write.table.type" -> tableType,
"hoodie.table.cdc.enabled" -> "true",
"hoodie.table.cdc.supplemental.logging.mode" -> "DATA_BEFORE_AFTER")
df.write.format("hudi").options(o).mode(SaveMode.Overwrite).save(path)
df.where("id = 1")
.withColumn("amt",
lit(java.math.BigDecimal.valueOf(9.99)).cast("decimal(7,2)"))
.withColumn("qty", lit(2))
.write.format("hudi").options(o + ("hoodie.datasource.write.operation"
-> "upsert"))
.mode("append").save(path)
spark.sql(s"select before, after from hudi_table_changes('$path', 'cdc',
'earliest') where op = 'u'")
.show(false)
}
run("COPY_ON_WRITE", "/tmp/cdc_types_cow")
run("MERGE_ON_READ", "/tmp/cdc_types_mor")
```
Actual output:
```
COPY_ON_WRITE before={"id": 1, "amt": [0, 0, 0, 125], "d": 20739, "ts":
1791865800000000, "qty": 1}
after ={"id": 1, "amt": [0, 0, 3, -25], "d": 20739, "ts":
1791865800000000, "qty": 2}
MERGE_ON_READ
before={"id":1,"amt":{"zero":false},"d":20739,"ts":1791865800000000,"qty":1}
after
={"id":1,"amt":{"zero":false},"d":20739,"ts":1791865800000000,"qty":2}
```
`from_json(after, <table schema>)` returns null for `amt` in both cases.
### Root cause
- **MOR:** `InternalRowToJsonStringConverter.convertField`
(`hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala`,
default branch at line 94 on `release-1.2.1`, line 130 on master) has no
`DecimalType` case.
Spark's internal `org.apache.spark.sql.types.Decimal` falls through to
Jackson, which
serializes it as a bean; its only bean-style getter is `isZero()`.
`DateType` and
`TimestampType` fall through as their internal `Int` and `Long`.
- **COW:** `HoodieCDCUtils.recordToJson`
(`hudi-common/.../table/cdc/HoodieCDCUtils.java`,
line 143 on `release-1.2.1`, line 160 on master) is
`GenericData.get().toString(record)`. That
prints Avro decimal logical types as their underlying bytes, and date and
timestamp logical
types as numbers.
Master added VARIANT handling to `convertField` (#19583) but nothing for
DECIMAL, DATE, or
TIMESTAMP.
### Suggested fix
Convert logical types to readable values in both paths:
- `convertField` adds cases for `DecimalType` (`Decimal.toJavaBigDecimal`),
`DateType`, and
`TimestampType`/`TimestampNTZType`.
- `recordToJson` uses a converter that applies Avro logical-type conversions
instead of
`GenericData.toString`.
Add a test with decimal, date, and timestamp columns on both table types,
asserting that the two
produce the same image.
### Environment
**Hudi version:** 1.2.1 (also reproduced by reading master's code path)
**Query engine:** Spark 3.5.6, Scala 2.12, Java 11
### Logs and Stack Trace
_No response_
--
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]
