Fix stats collection for integer based decimal numbers

Alchemy-item: (ID = 1247) Fix stats collection for integer based decimal numbers commit 1/1 - 8531b53

Author: mohsaka

velox/connectors/hive/iceberg/DataFileStatsCollector.cpp

@@ -38,12 +38,18 @@ void DataFileStatsCollector::collectStats(
   VELOX_CHECK_NOT_NULL(fileMetadata);
 
   std::unordered_set<int32_t> skipBoundsFields;
+  std::unordered_map<int32_t, TypePtr> fieldIdToType;
   std::function<int32_t(IcebergDataFileStatsSettings*)> processFields =
       [&skipBoundsFields,
+       &fieldIdToType,
        &processFields](IcebergDataFileStatsSettings* field) -> int32_t {
     if (field->skipBounds) {
       skipBoundsFields.insert(field->fieldId);
     }
+    // Store the Velox type for this field
+    if (field->veloxType) {
+      fieldIdToType[field->fieldId] = field->veloxType;
+    }
     if (field->children.empty()) {
       return 1;
     }
@@ -112,13 +118,29 @@ void DataFileStatsCollector::collectStats(
     }
   }
 
+  // Only 4-byte DECIMAL values (precision <= 9) need byte-swapping.
   for (const auto& [fieldId, minStats] : globalMinStats) {
-    const auto lowerBound = minStats->MinValue();
+    auto lowerBound = minStats->MinValue();
+    auto upperBound = globalMaxStats[fieldId]->MaxValue();
+
+    auto typeIt = fieldIdToType.find(fieldId);
+    if (typeIt != fieldIdToType.end() && typeIt->second &&
+        typeIt->second->isDecimal() && lowerBound.size() == 4) {
+      std::string lowerResult(4, '\0');
+      auto lowerSwapped = __builtin_bswap32(
+          *reinterpret_cast<const uint32_t*>(lowerBound.data()));
+      std::memcpy(lowerResult.data(), &lowerSwapped, 4);
+      lowerBound = lowerResult;
+
+      std::string upperResult(4, '\0');
+      auto upperSwapped = __builtin_bswap32(
+          *reinterpret_cast<const uint32_t*>(upperBound.data()));
+      std::memcpy(upperResult.data(), &upperSwapped, 4);
+      upperBound = upperResult;
+    }
+
     dataFileStats->lowerBounds[fieldId] =
         encoding::Base64::encode(lowerBound.data(), lowerBound.size());
-  }
-  for (const auto& [fieldId, maxStats] : globalMaxStats) {
-    const auto upperBound = maxStats->MaxValue();
     dataFileStats->upperBounds[fieldId] =
         encoding::Base64::encode(upperBound.data(), upperBound.size());
   }

velox/connectors/hive/iceberg/DataFileStatsCollector.h

@@ -16,6 +16,7 @@
 #pragma once
 
 #include "velox/dwio/common/DataFileStatsCollector.h"
+#include "velox/type/Type.h"
 
 namespace facebook::velox::connector::hive::iceberg {
 
@@ -26,10 +27,11 @@ struct IcebergDataFileStatsSettings
     : public dwio::common::DataFileStatsSettings {
   int32_t fieldId;
   bool skipBounds;
+  TypePtr veloxType;
   std::vector<std::unique_ptr<IcebergDataFileStatsSettings>> children;
 
-  IcebergDataFileStatsSettings(int32_t id, bool skip)
-      : fieldId(id), skipBounds(skip), children() {}
+  IcebergDataFileStatsSettings(int32_t id, bool skip, TypePtr type = nullptr)
+      : fieldId(id), skipBounds(skip), veloxType(type), children() {}
 };
 
 class DataFileStatsCollector : public dwio::common::FileStatsCollector {

velox/connectors/hive/iceberg/IcebergDataSink.cpp

@@ -221,7 +221,7 @@ IcebergDataSink::IcebergDataSink(
                              bool skipBounds) -> IcebergDataFileStatsSettings {
     VELOX_CHECK_NOT_NULL(type, "Input column type cannot be null.");
     bool currentSkipBounds = skipBounds || type->isMap() || type->isArray();
-    IcebergDataFileStatsSettings field(f.id, currentSkipBounds);
+    IcebergDataFileStatsSettings field(f.id, currentSkipBounds, type);
     if (!f.children.empty()) {
       VELOX_CHECK_EQ(f.children.size(), type->size());
       field.children.reserve(f.children.size());