Fix iceberg min max statistics for decimal type when encoded as int32

Signed-off-by: Hazmi <ialhazmim@gmail.com>

Alchemy-item: (ID = 1203) Fix iceberg min max statistics for decimal type when encoded as int32 commit 1/1 - 0ac9930

Author: infvg

velox/dwio/parquet/writer/arrow/Statistics.cpp

@@ -559,7 +559,9 @@ TypedComparatorImpl<false, ByteArrayType>::getMinMax(
 template <typename T>
 std::string encodeDecimalToBigEndian(T value) {
   uint8_t buffer[sizeof(T)];
-  if constexpr (std::is_same_v<T, int64_t>) {
+  if constexpr (std::is_same_v<T, int32_t>) {
+    *reinterpret_cast<int32_t*>(buffer) = ::arrow::bit_util::ToBigEndian(value);
+  } else if constexpr (std::is_same_v<T, int64_t>) {
     *reinterpret_cast<int64_t*>(buffer) = ::arrow::bit_util::ToBigEndian(value);
   } else if constexpr (std::is_same_v<T, int128_t>) {
     *reinterpret_cast<int128_t*>(buffer) = DecimalUtil::bigEndian(value);
@@ -809,6 +811,11 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
   }
 
   std::string MinValue() const override {
+    if constexpr (std::is_same_v<T, int32_t>) {
+      if (descr_->logicalType()->isDecimal()) {
+        return encodeDecimalToBigEndian(min_);
+      }
+    }
     if constexpr (std::is_same_v<T, int64_t>) {
       if (descr_->logicalType()->isDecimal()) {
         return encodeDecimalToBigEndian(min_);
@@ -835,6 +842,11 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
   }
 
   std::string MaxValue() const override {
+    if constexpr (std::is_same_v<T, int32_t>) {
+      if (descr_->logicalType()->isDecimal()) {
+        return encodeDecimalToBigEndian(max_);
+      }
+    }
     if constexpr (std::is_same_v<T, int64_t>) {
       if (descr_->logicalType()->isDecimal()) {
         return encodeDecimalToBigEndian(max_);

velox/dwio/parquet/writer/arrow/tests/StatisticsTest.cpp

@@ -2152,6 +2152,38 @@ TEST(IcebergStatistics, unboundedUpperBound) {
   }
 }
 
+TEST(IcebergStatistics, maxValueWithNulls) {
+  const NodePtr node = PrimitiveNode::make(
+      "decimal_col",
+      Repetition::kOptional,
+      LogicalType::decimal(7, 2),
+      Type::kInt32);
+  ColumnDescriptor descr(node, 1, 1);
+
+  auto stats = makeStatistics<Int32Type>(&descr);
+
+  std::vector<int32_t> values = {19900, 20000};
+  stats->update(values.data(), values.size(), 1);
+
+  ASSERT_TRUE(stats->hasMinMax());
+  EXPECT_EQ(stats->min(), 19900);
+  EXPECT_EQ(stats->max(), 20000);
+
+  const auto maxValue = stats->MaxValue();
+  EXPECT_FALSE(maxValue.empty()) << "MaxValue() should not be empty";
+
+  int32_t decodedMax = ::arrow::bit_util::FromBigEndian(
+      *reinterpret_cast<const int32_t*>(maxValue.data()));
+  EXPECT_EQ(decodedMax, 20000) << "MaxValue() should return 20000";
+
+  const auto minValue = stats->MinValue();
+  EXPECT_FALSE(minValue.empty()) << "MinValue() should not be empty";
+
+  int32_t decodedMin = ::arrow::bit_util::FromBigEndian(
+      *reinterpret_cast<const int32_t*>(minValue.data()));
+  EXPECT_EQ(decodedMin, 19900) << "MinValue() should return 19900";
+    }
+
 TEST(StatisticsComparison, withInt64) {
   NodePtr Node =
       PrimitiveNode::make("int_col", Repetition::kRequired, Type::kInt64);