perf(aggregate): use hash lookup for scalar field aggregators

src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp

@@ -19,7 +19,10 @@
 
 #include "paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h"
 
+#include <cassert>
+#include <cstdint>
 #include <cstring>
+#include <functional>
 #include <string_view>
 #include <utility>
 #include <vector>
@@ -34,6 +37,7 @@
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/common/utils/date_time_utils.h"
 #include "paimon/common/utils/fields_comparator.h"
+#include "paimon/common/utils/math.h"
 #include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/status.h"
@@ -123,6 +127,10 @@ Result<bool> EqualMaps(const std::shared_ptr<InternalMap>& lhs,
     return true;
 }
 
+size_t CombineHash(size_t lhs, size_t rhs) {
+    return lhs ^ (rhs + static_cast<size_t>(0x9e3779b97f4a7c15ULL) + (lhs << 6) + (lhs >> 2));
+}
+
 }  // namespace
 
 VariantType FieldAggregateUtils::OwnedBinary(const VariantType& value, MemoryPool* pool) {
@@ -244,4 +252,67 @@ Result<bool> FieldAggregateUtils::Equals(const VariantType& lhs, const VariantTy
     }
 }
 
+bool FieldAggregateUtils::IsHashableType(const std::shared_ptr<arrow::DataType>& type) {
+    switch (type->id()) {
+        case arrow::Type::BOOL:
+        case arrow::Type::INT8:
+        case arrow::Type::INT16:
+        case arrow::Type::INT32:
+        case arrow::Type::DATE32:
+        case arrow::Type::INT64:
+        case arrow::Type::FLOAT:
+        case arrow::Type::DOUBLE:
+        case arrow::Type::STRING:
+        case arrow::Type::BINARY:
+        case arrow::Type::TIMESTAMP:
+            return true;
+        default:
+            return false;
+    }
+}
+
+size_t FieldAggregateUtils::Hash(const VariantType& value,
+                                 const std::shared_ptr<arrow::DataType>& type) {
+    assert(IsHashableType(type));
+    size_t result = std::hash<int32_t>{}(static_cast<int32_t>(type->id()));
+    if (DataDefine::IsVariantNull(value)) {
+        return result;
+    }
+    switch (type->id()) {
+        case arrow::Type::BOOL:
+            return CombineHash(result, std::hash<bool>{}(DataDefine::GetVariantValue<bool>(value)));
+        case arrow::Type::INT8:
+            return CombineHash(result, std::hash<char>{}(DataDefine::GetVariantValue<char>(value)));
+        case arrow::Type::INT16:
+            return CombineHash(result,
+                               std::hash<int16_t>{}(DataDefine::GetVariantValue<int16_t>(value)));
+        case arrow::Type::INT32:
+        case arrow::Type::DATE32:
+            return CombineHash(result,
+                               std::hash<int32_t>{}(DataDefine::GetVariantValue<int32_t>(value)));
+        case arrow::Type::INT64:
+            return CombineHash(result,
+                               std::hash<int64_t>{}(DataDefine::GetVariantValue<int64_t>(value)));
+        case arrow::Type::FLOAT:
+            return CombineHash(result, std::hash<int32_t>{}(CanonicalizeFloatToIntBits(
+                                           DataDefine::GetVariantValue<float>(value))));
+        case arrow::Type::DOUBLE:
+            return CombineHash(result, std::hash<int64_t>{}(CanonicalizeDoubleToLongBits(
+                                           DataDefine::GetVariantValue<double>(value))));
+        case arrow::Type::STRING:
+        case arrow::Type::BINARY:
+            return CombineHash(result,
+                               std::hash<std::string_view>{}(DataDefine::GetStringView(value)));
+        case arrow::Type::TIMESTAMP: {
+            const auto& timestamp = DataDefine::GetVariantValue<Timestamp>(value);
+            return CombineHash(result,
+                               CombineHash(std::hash<int64_t>{}(timestamp.GetMillisecond()),
+                                           std::hash<int32_t>{}(timestamp.GetNanoOfMillisecond())));
+        }
+        default:
+            assert(false);
+            return 0;
+    }
+}
+
 }  // namespace paimon

src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h

@@ -19,6 +19,8 @@
 
 #pragma once
 
+#include <cassert>
+#include <cstddef>
 #include <memory>
 
 #include "arrow/type_fwd.h"
@@ -62,6 +64,36 @@ class FieldAggregateUtils {
     /// @return Whether the values are equal, or an error Status.
     static Result<bool> Equals(const VariantType& lhs, const VariantType& rhs,
                                const std::shared_ptr<arrow::DataType>& type);
+
+    /// Return whether values of this logical type have a hash compatible with Equals().
+    static bool IsHashableType(const std::shared_ptr<arrow::DataType>& type);
+
+    /// Hash a value according to the equality semantics of its logical type.
+    ///
+    /// The caller must first check IsHashableType(type).
+    static size_t Hash(const VariantType& value, const std::shared_ptr<arrow::DataType>& type);
 };
 
+namespace detail {
+
+struct SemanticHash {
+    std::shared_ptr<arrow::DataType> type;
+
+    size_t operator()(const VariantType& value) const {
+        return FieldAggregateUtils::Hash(value, type);
+    }
+};
+
+struct SemanticEqual {
+    std::shared_ptr<arrow::DataType> type;
+
+    bool operator()(const VariantType& lhs, const VariantType& rhs) const {
+        Result<bool> equal = FieldAggregateUtils::Equals(lhs, rhs, type);
+        assert(equal.ok());
+        return equal.value();
+    }
+};
+
+}  // namespace detail
+
 }  // namespace paimon

src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp

@@ -19,6 +19,7 @@
 
 #include "paimon/core/mergetree/compact/aggregate/field_collect_agg.h"
 
+#include <unordered_set>
 #include <vector>
 
 #include "arrow/api.h"
@@ -33,9 +34,12 @@
 namespace paimon {
 namespace {
 
-// TODO(liangjie.liang): Hash VariantType by its type so that this scan, the key lookup in
-// FieldMergeMapAgg and the keyed upsert in FieldNestedUpdateAgg stop being O(n^2). Java only pays
-// that cost for constructed element types and uses HashSet/HashMap for the rest.
+constexpr int32_t kHashThreshold = 192;
+using detail::SemanticEqual;
+using detail::SemanticHash;
+
+// Constructed element types stay on the linear path because their equality semantics are not
+// covered by FieldAggregateUtils::Hash.
 Result<bool> Contains(const std::vector<VariantType>& values, const VariantType& candidate,
                       const std::shared_ptr<arrow::DataType>& element_type) {
     for (const VariantType& value : values) {
@@ -68,6 +72,23 @@ Status AppendArray(const std::shared_ptr<InternalArray>& array,
     return Status::OK();
 }
 
+Status AppendArrayWithHash(const std::shared_ptr<InternalArray>& array,
+                           const std::shared_ptr<arrow::DataType>& element_type,
+                           std::unordered_set<VariantType, SemanticHash, SemanticEqual>* seen,
+                           std::vector<VariantType>* values) {
+    if (!array) {
+        return Status::OK();
+    }
+    for (int32_t i = 0; i < array->Size(); ++i) {
+        PAIMON_ASSIGN_OR_RAISE(VariantType value,
+                               FieldAggregateUtils::GetValue(*array, i, element_type));
+        if (seen->insert(value).second) {
+            values->push_back(std::move(value));
+        }
+    }
+    return Status::OK();
+}
+
 }  // namespace
 
 Result<std::unique_ptr<FieldCollectAgg>> FieldCollectAgg::Create(
@@ -111,12 +132,22 @@ Result<VariantType> FieldCollectAgg::AggImpl(const VariantType& accumulator,
     std::shared_ptr<InternalArray> input_array =
         input_null ? nullptr
                    : DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(input_field);
+    int32_t total_size = (accumulator_array ? accumulator_array->Size() : 0) +
+                         (input_array ? input_array->Size() : 0);
     std::vector<VariantType> values;
-    if (accumulator_array) {
-        values.reserve(accumulator_array->Size() + (input_array ? input_array->Size() : 0));
+    values.reserve(total_size);
+    if (distinct_ && total_size >= kHashThreshold &&
+        FieldAggregateUtils::IsHashableType(element_type_)) {
+        SemanticHash hasher{element_type_};
+        SemanticEqual equal{element_type_};
+        std::unordered_set<VariantType, SemanticHash, SemanticEqual> seen(0, hasher, equal);
+        seen.reserve(total_size);
+        PAIMON_RETURN_NOT_OK(AppendArrayWithHash(accumulator_array, element_type_, &seen, &values));
+        PAIMON_RETURN_NOT_OK(AppendArrayWithHash(input_array, element_type_, &seen, &values));
+    } else {
+        PAIMON_RETURN_NOT_OK(AppendArray(accumulator_array, element_type_, distinct_, &values));
+        PAIMON_RETURN_NOT_OK(AppendArray(input_array, element_type_, distinct_, &values));
     }
-    PAIMON_RETURN_NOT_OK(AppendArray(accumulator_array, element_type_, distinct_, &values));
-    PAIMON_RETURN_NOT_OK(AppendArray(input_array, element_type_, distinct_, &values));
     std::vector<std::shared_ptr<InternalArray>> holders;
     if (accumulator_array) {
         holders.push_back(accumulator_array);