fix: support non-string types for iceberg partition listing

Co-authored-by: Thomas Chow <[email protected]>

Author: tchow-zlai

spark/src/main/scala/ai/chronon/spark/format/Iceberg.scala

@@ -1,6 +1,7 @@
 package ai.chronon.spark.format
 
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions.{col, date_format}
 import org.apache.spark.sql.types.StructType
 
 case object Iceberg extends Format {
@@ -29,12 +30,12 @@ case object Iceberg extends Format {
       .load(s"$tableName.partitions")
 
     val index = partitionsDf.schema.fieldIndex("partition")
-
+    val partitionFmt = sparkSession.conf.get("spark.chronon.partition.format", "yyyyMMdd")
     if (partitionsDf.schema(index).dataType.asInstanceOf[StructType].fieldNames.contains("hr")) {
       // Hour filter is currently buggy in iceberg. https://github.com/apache/iceberg/issues/4718
       // so we collect and then filter.
       partitionsDf
-        .select("partition.ds", "partition.hr")
+        .select(date_format(col("partition.ds"), partitionFmt), col("partition.hr"))
         .collect()
         .filter(_.get(1) == null)
         .map(_.getString(0))
@@ -43,7 +44,7 @@ case object Iceberg extends Format {
     } else {
 
       partitionsDf
-        .select("partition.ds")
+        .select(date_format(col("partition.ds"), partitionFmt))
         .collect()
         .map(_.getString(0))
         .toSeq