Use unscaled value as partition value

Author: rui-mo

backends-velox/src/main/scala/io/glutenproject/backendsapi/velox/IteratorApiImpl.scala

@@ -25,7 +25,6 @@ import io.glutenproject.substrait.rel.{LocalFilesBuilder, LocalFilesNode, SplitI
 import io.glutenproject.substrait.rel.LocalFilesNode.ReadFileFormat
 import io.glutenproject.utils._
 import io.glutenproject.vectorized._
-
 import org.apache.spark.{SparkConf, SparkContext, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
@@ -35,7 +34,7 @@ import org.apache.spark.sql.catalyst.util.{DateFormatter, TimestampFormatter}
 import org.apache.spark.sql.connector.read.InputPartition
 import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
 import org.apache.spark.sql.execution.metric.SQLMetric
-import org.apache.spark.sql.types.{BinaryType, DateType, StructType, TimestampType}
+import org.apache.spark.sql.types.{BinaryType, DateType, Decimal, DecimalType, StructType, TimestampType}
 import org.apache.spark.sql.utils.OASPackageBridge.InputMetricsWrapper
 import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.util.ExecutorManager
@@ -45,7 +44,6 @@ import java.nio.charset.StandardCharsets
 import java.time.ZoneOffset
 import java.util.{ArrayList => JArrayList, HashMap => JHashMap, Map => JMap}
 import java.util.concurrent.TimeUnit
-
 import scala.collection.JavaConverters._
 
 class IteratorApiImpl extends IteratorApi with Logging {
@@ -118,6 +116,8 @@ class IteratorApiImpl extends IteratorApi with Logging {
                 new String(pn.asInstanceOf[Array[Byte]], StandardCharsets.UTF_8)
               case _: DateType =>
                 DateFormatter.apply().format(pn.asInstanceOf[Integer])
+              case _: DecimalType =>
+                pn.asInstanceOf[Decimal].toUnscaledLong.toString
               case _: TimestampType =>
                 TimestampFormatter
                   .getFractionFormatter(ZoneOffset.UTC)