rangareddy commented on issue #15970:
URL: https://github.com/apache/hudi/issues/15970#issuecomment-4889969014
**Sample Tested code:**
```python
import os
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType,
BooleanType
root = os.environ.get("ROOT", "/tmp/cs15970py")
spark = SparkSession.builder.appName("hudi-15970").getOrCreate()
spark.sparkContext.setLogLevel("WARN")
base_schema = StructType([
StructField("id", StringType(), False),
StructField("ts", LongType(), False),
StructField("part", StringType(), False),
StructField("fare", LongType(), False),
])
del_schema = StructType(base_schema.fields +
[StructField("_hoodie_is_deleted", BooleanType(), False)])
def opts(extra=None):
o = {
"hoodie.table.name": "cs_mor",
"hoodie.datasource.write.recordkey.field": "id",
"hoodie.datasource.write.precombine.field": "ts",
"hoodie.datasource.write.partitionpath.field": "part",
"hoodie.datasource.write.table.type": "MERGE_ON_READ",
"hoodie.metadata.enable": "true",
"hoodie.metadata.index.column.stats.enable": "true",
"hoodie.metadata.index.column.stats.column.list": "fare",
"hoodie.datasource.write.hive_style_partitioning": "true",
}
if extra:
o.update(extra)
return o
def wr(rows, schema, o, op, mode, path):
df = spark.createDataFrame(rows, schema) # rows are position-matched
tuples
(df.write.format("hudi").options(**o)
.option("hoodie.datasource.write.operation",
op).mode(mode).save(path))
def rd(path, pred):
return (spark.read.format("hudi")
.option("hoodie.metadata.enable", "true")
.option("hoodie.enable.data.skipping", "true")
.load(path).where(pred).count())
def dump_cs(path):
try:
(spark.read.format("hudi").load(path + "/.hoodie/metadata")
.where("type = 3 AND ColumnStatsMetadata.columnName = 'fare'")
.selectExpr("ColumnStatsMetadata.fileName AS file",
"ColumnStatsMetadata.minValue AS minV",
"ColumnStatsMetadata.maxValue AS maxV",
"ColumnStatsMetadata.nullCount AS nulls",
"ColumnStatsMetadata.isDeleted AS
del").show(truncate=False))
except Exception as e:
print(">>> (col_stats dump skipped: %s)" % type(e).__name__)
# ---- items (1)+(2): one value per slice, merged base+log ----
try:
p = root + "/slice"
wr([("a", 1, "p1", 10), ("b", 1, "p1", 20), ("c", 1, "p1", 30)],
base_schema, opts(), "insert", "overwrite", p)
wr([("a", 2, "p1", 1000)], base_schema, opts(), "upsert", "append", p)
# fare=1000 lives ONLY in the
total, log_only, beyond = rd(p, "1=1"), rd(p, "fare >= 900"), rd(p,
"fare > 5000")
print(">>> SLICE col_stats (one row per data file in the slice):");
dump_cs(p)
ok = (total == 3 and log_only == 1 and beyond == 0)
print(">>> RESULT=SLICE %s total=%d fare>=900=%d fare>5000=%d (expect 3,
1 [log value survives], 0)"
% ("OK" if ok else "MISMATCH", total, log_only, beyond))
except Exception as e:
print(">>> RESULT=SLICE ERROR :: %s :: %s" % (type(e).__name__,
str(e)[:200]))
# ---- item (3): delete log block (write.operation=delete) ----
try:
p = root + "/del"
wr([("a", 1, "p1", 10), ("b", 1, "p1", 20), ("c", 1, "p1", 30), ("d", 1,
"p1", 1000)],
base_schema, opts(), "insert", "overwrite", p)
wr([("d", 2, "p1", 1000)], base_schema, opts(), "delete", "append", p)
total, hi = rd(p, "1=1"), rd(p, "fare >= 900")
print(">>> DELETE_BLOCK col_stats:"); dump_cs(p)
print(">>> RESULT=DELETE_BLOCK %s total=%d fare>=900=%d (expect 3, 0 ->
d deleted, no stale read)"
% ("OK" if (total == 3 and hi == 0) else "MISMATCH", total, hi))
except Exception as e:
print(">>> RESULT=DELETE_BLOCK ERROR :: %s :: %s" % (type(e).__name__,
str(e)[:200]))
# ---- item (4): custom delete payload (soft delete via _hoodie_is_deleted)
----
try:
p = root + "/custdel"
o = opts({"hoodie.datasource.write.payload.class":
"org.apache.hudi.common.model.DefaultHoodieRecordPayload"})
wr([("a", 1, "p1", 10, False), ("b", 1, "p1", 20, False), ("c", 1, "p1",
30, False), ("d", 1, "p1", 1000, False)],
del_schema, o, "insert", "overwrite", p)
wr([("d", 2, "p1", 1000, True)], del_schema, o, "upsert", "append", p)
total, hi = rd(p, "1=1"), rd(p, "fare >= 900")
print(">>> CUSTOM_PAYLOAD col_stats:"); dump_cs(p)
print(">>> RESULT=CUSTOM_PAYLOAD %s total=%d fare>=900=%d (expect 3, 0
-> d soft-deleted via payload)"
% ("OK" if (total == 3 and hi == 0) else "MISMATCH", total, hi))
except Exception as e:
print(">>> RESULT=CUSTOM_PAYLOAD ERROR :: %s :: %s" % (type(e).__name__,
str(e)[:200]))
spark.stop()
```
**Output:**
```sh
>>> SLICE col_stats (one row per data file in the slice):
+-------------------------------------------------------------------------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|file
|minV
|maxV
|nulls|del |
+-------------------------------------------------------------------------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|.ce0f7c3c-03a4-460f-9993-c18c61e238f1-0_20260706065436335.log.1_0-68-161
|{null, null, {1000}, null, null, null, null, null, null, null, null, null,
null}|{null, null, {1000}, null, null, null, null, null, null, null, null,
null, null}|0 |false|
|ce0f7c3c-03a4-460f-9993-c18c61e238f1-0_0-44-101_20260706065429746.parquet|{null,
null, {10}, null, null, null, null, null, null, null, null, null, null}
|{null, null, {30}, null, null, null, null, null, null, null, null, null, null}
|0 |false|
+-------------------------------------------------------------------------+--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
>>> RESULT=SLICE OK total=3 fare>=900=1 fare>5000=0 (expect 3, 1 [log value
survives], 0)
>>> DELETE_BLOCK col_stats:
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|file
|minV
|maxV
|nulls|del |
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|a0e304d1-053f-40f3-a18f-8524a2bdd29e-0_0-122-261_20260706065442514.parquet|{null,
null, {10}, null, null, null, null, null, null, null, null, null, null}|{null,
null, {1000}, null, null, null, null, null, null, null, null, null, null}|0
|false|
|.a0e304d1-053f-40f3-a18f-8524a2bdd29e-0_20260706065445618.log.1_0-150-322
|null
|null
|0 |false|
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
>>> CUSTOM_PAYLOAD col_stats:
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|file
|minV
|maxV
|nulls|del |
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
|cc6dedf6-dd0e-402a-94e5-10c538641e32-0_0-198-417_20260706065449261.parquet|{null,
null, {10}, null, null, null, null, null, null, null, null, null, null}|{null,
null, {1000}, null, null, null, null, null, null, null, null, null, null}|0
|false|
|.cc6dedf6-dd0e-402a-94e5-10c538641e32-0_20260706065452291.log.1_0-222-477
|null
|null
|0 |false|
+--------------------------------------------------------------------------+------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-----+-----+
>>> RESULT=CUSTOM_PAYLOAD OK total=3 fare>=900=0 (expect 3, 0 -> d
soft-deleted via payload)
```
--
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]
