Re: [I] Fix timestamp based partitioning and drop partition support with 0.15.0 [hudi]

["rangareddy (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org