rangareddy commented on issue #16679:
URL: https://github.com/apache/hudi/issues/16679#issuecomment-4892010614
**Sample Reproducible Code:**
```python
from pyspark.sql.types import StructType, StructField, StringType,
IntegerType, LongType
table_name = "vk_multikey"
table_path = f"s3://warehouse/{table_name}"
schema = StructType([
StructField("k1", StringType(), False),
StructField("k2", IntegerType(), False),
StructField("name", StringType(), True),
StructField("ts", LongType(), False),
StructField("part", StringType(), False),
StructField("fare", LongType(), True),
])
opts = {
"hoodie.table.name": table_name,
"hoodie.datasource.write.recordkey.field": "k1,k2", #
MULTIPLE record key fields
"hoodie.datasource.write.partitionpath.field": "part",
"hoodie.datasource.write.precombine.field": "ts",
"hoodie.datasource.write.keygenerator.class":
"org.apache.hudi.keygen.ComplexKeyGenerator",
"hoodie.datasource.write.table.type": "MERGE_ON_READ",
"hoodie.populate.meta.fields": "false", #
virtual keys
"hoodie.metadata.enable": "false",
"hoodie.datasource.write.hive_style_partitioning": "true",
}
def wr(rows, op, mode):
(spark.createDataFrame(rows, schema).write.format("hudi").options(**opts)
.option("hoodie.datasource.write.operation",
op).mode(mode).save(table_path))
# base file: composite keys (a,1),(a,2),(b,1)
wr([("a", 1, "alice", 1, "p1", 10), ("a", 2, "bob", 1, "p1", 20), ("b", 1,
"carol", 1, "p1", 30)],
"insert", "overwrite")
# log update on composite key (a,1) -> name=alice2, fare=100 (must merge
onto the base record)
wr([("a", 1, "alice2", 2, "p1", 100)], "upsert", "append")
try:
df = spark.read.format("hudi").load(table_path) # snapshot read
-> file group reader
total = df.count()
dup = df.groupBy("k1", "k2").count().where("count > 1").count()
a1 = df.where("k1='a' AND k2=1").select("name", "fare").collect()
print(">>> total=%d dup_composite_keys=%d (a,1)=%s" % (total, dup,
[(r['name'], r['fare']) for r in a1]))
ok = (total == 3 and dup == 0 and len(a1) == 1 and a1[0]["name"] ==
"alice2" and a1[0]["fare"] == 100)
print(">>> RESULT=%s (expect total=3, no dup composite keys, (a,1)
updated to alice2/100)"
% ("PASS" if ok else "FAIL"))
except Exception as e:
print(">>> RESULT=FAILED :: %s :: %s" % (type(e).__name__, str(e)[:300]))
```
**Hudi 0.14 Output:**
```java
>>> RESULT=FAILED :: Py4JJavaError :: An error occurred while calling
o221.count.
: java.lang.IllegalStateException
at
org.apache.hudi.common.util.ValidationUtils.checkState(ValidationUtils.java:62)
at
org.apache.hudi.HoodieBaseRelation.recordKeyField$lzycompute(HoodieBaseRelation.scala:129)
```
**Hudi 1.2.0 Output:**
```java
>>> total=3 dup_composite_keys=0 (a,1)=[('alice2', 100)]
>>> RESULT=PASS (expect total=3, no dup composite keys, (a,1) updated to
alice2/100)
```
--
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]
