apache / apache/hudi

CDC data_before_after mode does not convert Spark DecimalType correctly

Open
#8,616 7 comments 0 reactions 0 assignees View on GitHub
area:schema engine:spark
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

I've noticed an issue with the `data_before_after` CDC mode not converting Spark DecimalType correctly. The decimals are getting converted to an array in the before and after json strings when the cdc data is saved, which then results in null values when trying to convert back to a Row using F.from_json() along with the original schema because Spark can't cast the array to a valid DecimalType. Example:

Querying the hudi table normally:
`gljeln=Decimal('208.000000000000000000')`

Querying using the cdc format + incremental options:
`Row(op='i', ts_ms='20230425193451991', before='null', after='{"gljeln": [0, 0, 0, 0, 0, 0, 0, 11, 70, -108, 113, -8, 1, 64, 0, 0]...`

Steps to reproduce the behavior:

1. Create a Spark dataframe that contains a decimal column (precision=38, scale=18) with valid decimal value.
2. Write dataframe to new Hudi table using write options for enabling CDC.
3. Create a second dataframe that contains a decimal column (precision=38, scale=18) with valid decimal value.
4. Write dataframe to same Hudi table using same write options, which will create the first .cdc file.
5. Read CDC data using cdc format for incremental query starting at the first commit time (which will only read the newly created cdc data).
6. The result is a single insert entry, and the decimal value in the "after" JSON string will be an array of numbers instead of the string representation of the decimal.

**Expected behavior**

The decimal value should be serialized to an appropriate type (probably a string) that can be deserialized without corrupting the data.

**Environment Description**

* Hudi version : hudi-spark3.3-bundle_2.12-0.13.0.jar

* Spark version : 3.3.1

* Hive version : N/A

* Hadoop version : N/A

* Storage (HDFS/S3/GCS..) : S3

* Running on Docker? (yes/no) : Yes

**Additional context**

Example script to reproduce, and results:
```
from datetime import datetime
from decimal import Decimal
from pyspark.sql import SparkSession, types as T

HUDI_TARGET = 's3://some-bucket'

def decimal_test():
spark = (
SparkSession
.builder
.appName('decimal_test')
.config('spark.serializer', 'org.apache.spark.serializer.KryoSerializer')
.config('spark.sql.hive.convertMetastoreParquet', 'false')
.config('spark.hadoop.mapreduce.input.pathFilter.class', 'org.apache.hudi.hadoop.HoodieROTablePathFilter')
.config('spark.sql.parquet.mergeSchema', 'true')
.config('spark.sql.files.ignoreMissingFiles', 'true')
.config('spark.sql.adaptive.enabled', 'true')
.config('spark.sql.sources.partitionOverwriteMode', 'dynamic')
.config('spark.sql.sources.partitionColumnTypeInference.enabled', 'false')
.getOrCreate()
)
write_options = {
'hoodie.bloom.index.bucketized.checking': False,
'hoodie.bloom.index.input.storage.level': 'MEMORY_AND_DISK',
'hoodie.bloom.index.prune.by.ranges': False,
'hoodie.bulkinsert.shuffle.parallelism': 50,
'hoodie.bulkinsert.sort.mode': 'PARTITION_SORT',
'hoodie.combine.before.insert': False,
'hoodie.combine.before.upsert': False,
'hoodie.datasource.write.insert.drop.duplicates': False,
'hoodie.datasource.write.operation': 'UPSERT',
'hoodie.datasource.write.payload.class': 'org.apache.hudi.common.model.DefaultHoodieRecordPayload',
'hoodie.datasource.write.precombine.field': 'ts',
'hoodie.datasource.write.recordkey.field': 'id',
'hoodie.datasource.write.row.writer.enable': True,
'hoodie.datasource.write.table.type': 'COPY_ON_WRITE',
'hoodie.finalize.write.parallelism': 50,
'hoodie.index.type': 'SIMPLE',
'hoodie.insert.shuffle.parallelism': 50,
'hoodie.metadata.enable': True,
'hoodie.payload.ordering.field': 'ts',
'hoodie.simple.index.input.storage.level': 'MEMORY_AND_DISK',
'hoodie.table.name': 'some_table',
'hoodie.upsert.shuffle.parallelism': 50,
'hoodie.write.status.storage.level': 'MEMORY_AND_DISK',
'hoodie.table.cdc.enabled': True,
'hoodie.datasource.query.incremental.format': 'data_before_after',
'hoodie.datasource.write.partitionpath.field': '',
'hoodie.datasource.write.keygenerator.class': 'org.apache.hudi.keygen.NonpartitionedKeyGenerator'
}
schema = T.StructType([
T.StructField('id', T.IntegerType()),
T.StructField('amount', T.DecimalType(38, 18)),
T.StructField('ts', T.TimestampType())
])
df1 = spark.createDataFrame([(1, Decimal('42.000000000000000000'), datetime.now())], schema)
df2 = spark.createDataFrame([(2, Decimal('2319.000000000000000000'), datetime.now())], schema)
df1.write.format('hudi').options(**write_options).save(HUDI_TARGET, mode='append')
df1 = spark.read.format('hudi').load(HUDI_TARGET)
assert df1.first().amount == Decimal('42.000000000000000000')
df2.write.format('hudi').options(**write_options).save(HUDI_TARGET, mode='append')

if __name__ == '__main__':
decimal_test()
```

Incremental CDC query:
```
incremental_read_options = {
'hoodie.datasource.query.type': 'incremental',
'hoodie.datasource.read.begin.instanttime': commit_time,
'hoodie.datasource.query.incremental.format': 'cdc'
}

latest = spark.read.format('hudi').options(**incremental_read_options).load(s3_path)
```

And the result:
```
Row(op='i', ts_ms='20230428175640557', before='null', after='{"id": 2, "amount": [0, 0, 0, 0, 0, 0, 0, 125, -74, -105, 5, 105, 82, -36, 0, 0], "ts": 1682704564405033}')
```

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the provided Python reproduction and the CDC incremental query using the data_before_after format, then trace the DecimalType value through CDC serialization and the F.from_json() deserialization path. Done means the decimal is represented in a deserializable form and the reproduction preserves its value instead of producing an array or null.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, python
Domain
data-engineering, databases, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.