rangareddy commented on issue #17365:
URL: https://github.com/apache/hudi/issues/17365#issuecomment-4892474696
**Sample Reproducible Code:**
```python
table_name = "ts_multi_table"
table_path = f"s3://warehouse/{table_name}"
from pyspark.sql.types import StructType, StructField, IntegerType,
StringType, DoubleType, LongType
opts = {
"hoodie.datasource.write.table.type": "COPY_ON_WRITE",
"hoodie.datasource.write.keygenerator.class":
"org.apache.hudi.keygen.CustomKeyGenerator",
"hoodie.datasource.write.partitionpath.field":
"ts:timestamp,segment:simple", # multi-field, timestamp + simple
"hoodie.datasource.write.recordkey.field": "id",
"hoodie.datasource.write.precombine.field": "name",
"hoodie.table.name": table_name,
"hoodie.keygen.timebased.timestamp.type": "SCALAR",
"hoodie.keygen.timebased.output.dateformat": "yyyyMM",
"hoodie.keygen.timebased.timestamp.scalar.time.unit": "seconds",
"hoodie.metadata.enable": "false",
}
schema = StructType([
StructField("id", IntegerType(), False), StructField("name",
StringType(), False),
StructField("price", DoubleType(), False), StructField("ts", LongType(),
False),
StructField("segment", StringType(), False)])
rows = [(1, "a1", 1.6, 1704121827, "cat1"), (2, "a2", 10.8, 1704121827,
"cat1"),
(3, "a3", 30.0, 1706800227, "cat1"), (4, "a4", 103.4, 1701443427,
"cat2"),
(5, "a5", 1999.0, 1704121827, "cat2"), (6, "a6", 80.0, 1704121827,
"cat3")]
df = spark.createDataFrame(rows, schema)
INPUT_TS = sorted({1704121827, 1706800227, 1701443427})
try:
df.write.format("hudi").options(**opts).option("hoodie.datasource.write.operation",
"insert") \
.mode("overwrite").save(table_path)
r1 = spark.read.format("hudi").load(table_path)
ts_vals = sorted({row["ts"] for row in r1.select("ts").collect()})
parts = sorted({row["_hoodie_partition_path"] for row in
r1.select("_hoodie_partition_path").collect()})
print(">>> ts values read = %s expected original epochs = %s" %
(ts_vals, INPUT_TS))
print(">>> partitions = %s" % parts)
ts_ok = (ts_vals == INPUT_TS) # (A) data value preserved, not replaced
by partition value
# (B) drop one partition and confirm it disappears
drop_part = parts[0]
before = r1.count()
df.write.format("hudi").options(**opts) \
.option("hoodie.datasource.write.operation", "delete_partition") \
.option("hoodie.datasource.write.partitions.to.delete", drop_part) \
.mode("append").save(table_path)
r2 = spark.read.format("hudi").load(table_path)
after = r2.count()
remaining = sorted({row["_hoodie_partition_path"] for row in
r2.select("_hoodie_partition_path").collect()})
print(">>> dropped=%s before=%d after=%d remaining=%s" % (drop_part,
before, after, remaining))
drop_ok = (drop_part not in remaining) and (after < before)
print(">>> ts_value_correct=%s drop_partition_ok=%s %s"
% (ts_ok, drop_ok, "PASS" if (ts_ok and drop_ok) else "FAIL"))
except Exception as e:
print(">>> RESULT=FAILED :: %s :: %s" % (type(e).__name__, str(e)[:300]))
```
**Hudi 0.15.0 Output:**
```sh
>>> ts values read = [202312, 202401, 202402] expected original epochs =
[1701443427, 1704121827, 1706800227]
>>> partitions = ['202312/cat2', '202401/cat1', '202401/cat2',
'202401/cat3', '202402/cat1']
26/07/06 11:45:47 WARN HoodieSparkSqlWriterInternal: Closing write client
>>> dropped=202312/cat2 before=6 after=5 remaining=['202401/cat1',
'202401/cat2', '202401/cat3', '202402/cat1']
>>> ts_value_correct=False drop_partition_ok=True FAIL
```
**Hudi 1.2.0 Output:**
```sh
>>> ts values read = [1701443427, 1704121827, 1706800227] expected
original epochs = [1701443427, 1704121827, 1706800227]
>>> partitions = ['202312/cat2', '202401/cat1', '202401/cat2',
'202401/cat3', '202402/cat1']
>>> dropped=202312/cat2 before=6 after=5 remaining=['202401/cat1',
'202401/cat2', '202401/cat3', '202402/cat1']
>>> ts_value_correct=True drop_partition_ok=True PASS
```
--
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]
