diff --git a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
index 22e02f108..8dc4b49ff 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
+++ b/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
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h
index e3f74ed7e..db03de558 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h
+++ b/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
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp b/src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp
index ff938b73d..919286b70 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp
+++ b/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);
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_collect_agg_test.cpp b/src/paimon/core/mergetree/compact/aggregate/field_collect_agg_test.cpp
index 338045673..22389696f 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_collect_agg_test.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_collect_agg_test.cpp
@@ -20,6 +20,8 @@
 #include "paimon/core/mergetree/compact/aggregate/field_collect_agg.h"
 
 #include <algorithm>
+#include <cmath>
+#include <cstring>
 #include <memory>
 #include <string>
 #include <utility>
@@ -32,6 +34,7 @@
 #include "paimon/common/data/generic_row.h"
 #include "paimon/common/data/serializer/binary_serializer_utils.h"
 #include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/fields_comparator.h"
 #include "paimon/core/core_options.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/testing/utils/testharness.h"
@@ -60,6 +63,18 @@ Result<std::unique_ptr<FieldCollectAgg>> MakeCollectAgg(bool distinct) {
     return FieldCollectAgg::Create(arrow::list(arrow::int32()), options, "f", GetDefaultPool());
 }
 
+Result<std::unique_ptr<FieldCollectAgg>> MakeDistinctAgg(
+    const std::shared_ptr<arrow::DataType>& element_type) {
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
+                           CoreOptions::FromMap({{"fields.f.distinct", "true"}}));
+    return FieldCollectAgg::Create(arrow::list(element_type), options, "f", GetDefaultPool());
+}
+
+VariantType Array(std::vector<VariantType> values) {
+    return VariantType(
+        checked_pointer_cast<InternalArray>(std::make_shared<GenericArray>(std::move(values))));
+}
+
 }  // namespace
 
 TEST(FieldCollectAggTest, ConcatenatesWithoutReversing) {
@@ -94,6 +109,166 @@ TEST(FieldCollectAggTest, DistinctAndRetractOneOccurrence) {
     ASSERT_EQ((std::vector<int32_t>{1, 2, 3}), Values(retract_result));
 }
 
+TEST(FieldCollectAggTest, HashPathPreservesDuplicatesOverlapAndOrder) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> agg, MakeCollectAgg(true));
+    std::vector<VariantType> accumulator;
+    std::vector<VariantType> input;
+    for (int32_t i = 0; i < 100; ++i) {
+        accumulator.emplace_back(i % 40);
+        input.emplace_back(20 + (i % 40));
+    }
+
+    ASSERT_OK_AND_ASSIGN(VariantType result,
+                         agg->Agg(IntArray(std::move(accumulator)), IntArray(std::move(input))));
+    std::vector<int32_t> expected;
+    for (int32_t i = 0; i < 60; ++i) {
+        expected.push_back(i);
+    }
+    ASSERT_EQ(expected, Values(result));
+}
+
+TEST(FieldCollectAggTest, HashPathHandlesNullsAndStringContent) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> int_agg, MakeCollectAgg(true));
+    std::vector<VariantType> accumulator;
+    std::vector<VariantType> input;
+    for (int32_t i = 0; i < 100; ++i) {
+        accumulator.emplace_back(i == 0 ? VariantType(NullType()) : VariantType(int32_t{1}));
+        input.emplace_back(i == 0 ? VariantType(NullType()) : VariantType(int32_t{2}));
+    }
+    ASSERT_OK_AND_ASSIGN(VariantType int_result, int_agg->Agg(IntArray(std::move(accumulator)),
+                                                              IntArray(std::move(input))));
+    auto int_array = DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(int_result);
+    ASSERT_EQ(3, int_array->Size());
+    ASSERT_TRUE(int_array->IsNullAt(0));
+    ASSERT_EQ(1, int_array->GetInt(1));
+    ASSERT_EQ(2, int_array->GetInt(2));
+
+    for (const auto& element_type : {arrow::utf8(), arrow::binary()}) {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> string_agg,
+                             MakeDistinctAgg(element_type));
+        std::vector<std::string> left(100, "same-content");
+        std::vector<std::string> right(100, "same-content");
+        std::vector<VariantType> left_values;
+        std::vector<VariantType> right_values;
+        for (const auto& value : left) {
+            left_values.emplace_back(std::string_view(value));
+        }
+        for (const auto& value : right) {
+            right_values.emplace_back(std::string_view(value));
+        }
+        ASSERT_OK_AND_ASSIGN(
+            VariantType string_result,
+            string_agg->Agg(IntArray(std::move(left_values)), IntArray(std::move(right_values))));
+        auto string_array =
+            DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(string_result);
+        ASSERT_EQ(1, string_array->Size());
+        ASSERT_EQ("same-content", string_array->GetStringView(0));
+    }
+}
+
+TEST(FieldCollectAggTest, HashPathPreservesFloatingPointAndTimestampSemantics) {
+    uint32_t float_bits1 = 0x7fc00001U;
+    uint32_t float_bits2 = 0x7fc00002U;
+    uint64_t double_bits1 = 0x7ff8000000000001ULL;
+    uint64_t double_bits2 = 0x7ff8000000000002ULL;
+    float float_nan1;
+    float float_nan2;
+    double double_nan1;
+    double double_nan2;
+    std::memcpy(&float_nan1, &float_bits1, sizeof(float_nan1));
+    std::memcpy(&float_nan2, &float_bits2, sizeof(float_nan2));
+    std::memcpy(&double_nan1, &double_bits1, sizeof(double_nan1));
+    std::memcpy(&double_nan2, &double_bits2, sizeof(double_nan2));
+
+    auto verify_floating_point = [&](const std::shared_ptr<arrow::DataType>& type,
+                                     const VariantType& negative_zero, const VariantType& nan1,
+                                     const VariantType& positive_zero, const VariantType& nan2) {
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> agg, MakeDistinctAgg(type));
+        std::vector<VariantType> accumulator;
+        std::vector<VariantType> input;
+        for (int32_t i = 0; i < 100; ++i) {
+            accumulator.emplace_back(i % 2 == 0 ? negative_zero : nan1);
+            input.emplace_back(i % 2 == 0 ? positive_zero : nan2);
+        }
+        ASSERT_OK_AND_ASSIGN(VariantType result, agg->Agg(IntArray(std::move(accumulator)),
+                                                          IntArray(std::move(input))));
+        auto array = DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(result);
+        ASSERT_EQ(3, array->Size());
+        if (type->id() == arrow::Type::FLOAT) {
+            ASSERT_EQ(0, FieldsComparator::CompareFloatingPoint(array->GetFloat(0), -0.0F));
+            ASSERT_TRUE(std::isnan(array->GetFloat(1)));
+            ASSERT_EQ(0, FieldsComparator::CompareFloatingPoint(array->GetFloat(2), +0.0F));
+        } else {
+            ASSERT_EQ(0, FieldsComparator::CompareFloatingPoint(array->GetDouble(0), -0.0));
+            ASSERT_TRUE(std::isnan(array->GetDouble(1)));
+            ASSERT_EQ(0, FieldsComparator::CompareFloatingPoint(array->GetDouble(2), +0.0));
+        }
+    };
+    verify_floating_point(arrow::float32(), VariantType(-0.0F), VariantType(float_nan1),
+                          VariantType(+0.0F), VariantType(float_nan2));
+    verify_floating_point(arrow::float64(), VariantType(-0.0), VariantType(double_nan1),
+                          VariantType(+0.0), VariantType(double_nan2));
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> timestamp_agg,
+                         MakeDistinctAgg(arrow::timestamp(arrow::TimeUnit::NANO)));
+    Timestamp first = Timestamp::FromEpochMillis(100, 7);
+    Timestamp second = Timestamp::FromEpochMillis(101, 0);
+    std::vector<VariantType> timestamps;
+    std::vector<VariantType> more_timestamps;
+    for (int32_t i = 0; i < 100; ++i) {
+        timestamps.emplace_back(first);
+        more_timestamps.emplace_back(i == 0 ? first : second);
+    }
+    ASSERT_OK_AND_ASSIGN(
+        VariantType timestamp_result,
+        timestamp_agg->Agg(IntArray(std::move(timestamps)), IntArray(std::move(more_timestamps))));
+    auto timestamp_array =
+        DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(timestamp_result);
+    ASSERT_EQ(2, timestamp_array->Size());
+    ASSERT_EQ(first, timestamp_array->GetTimestamp(0, 9));
+    ASSERT_EQ(second, timestamp_array->GetTimestamp(1, 9));
+}
+
+TEST(FieldCollectAggTest, FallbackDecimalPathPreservesEquality) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> agg,
+                         MakeDistinctAgg(arrow::decimal128(10, 2)));
+    Decimal one = Decimal::FromUnscaledLong(100, 10, 2);
+    Decimal same_value = Decimal::FromUnscaledLong(1, 10, 0);
+    Decimal two = Decimal::FromUnscaledLong(200, 10, 2);
+    std::vector<VariantType> accumulator;
+    std::vector<VariantType> input;
+    for (int32_t i = 0; i < 100; ++i) {
+        accumulator.emplace_back(one);
+        input.emplace_back(i == 0 ? same_value : two);
+    }
+    ASSERT_OK_AND_ASSIGN(VariantType result,
+                         agg->Agg(IntArray(std::move(accumulator)), IntArray(std::move(input))));
+    auto array = DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(result);
+    ASSERT_EQ(2, array->Size());
+    ASSERT_EQ(one, array->GetDecimal(0, 10, 2));
+    ASSERT_EQ(two, array->GetDecimal(1, 10, 2));
+}
+
+TEST(FieldCollectAggTest, FallbackConstructedPathPreservesEquality) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldCollectAgg> agg,
+                         MakeDistinctAgg(arrow::list(arrow::int32())));
+    std::vector<VariantType> accumulator;
+    std::vector<VariantType> input;
+    for (int32_t i = 0; i < 100; ++i) {
+        accumulator.emplace_back(Array({int32_t{1}, int32_t{2}}));
+        input.emplace_back(i == 0 ? Array({int32_t{1}, int32_t{2}})
+                                  : Array({int32_t{2}, int32_t{3}}));
+    }
+    ASSERT_OK_AND_ASSIGN(VariantType result,
+                         agg->Agg(IntArray(std::move(accumulator)), IntArray(std::move(input))));
+    auto array = DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(result);
+    ASSERT_EQ(2, array->Size());
+    ASSERT_EQ(1, array->GetArray(0)->GetInt(0));
+    ASSERT_EQ(2, array->GetArray(0)->GetInt(1));
+    ASSERT_EQ(2, array->GetArray(1)->GetInt(0));
+    ASSERT_EQ(3, array->GetArray(1)->GetInt(1));
+}
+
 TEST(FieldCollectAggTest, RejectsNonArrayType) {
     ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
     ASSERT_NOK(FieldCollectAgg::Create(arrow::int32(), options, "f", GetDefaultPool()));
@@ -103,18 +278,6 @@ TEST(FieldCollectAggTest, RejectsNonArrayType) {
 // collection over composite element types.
 namespace {
 
-Result<std::unique_ptr<FieldCollectAgg>> MakeDistinctAgg(
-    const std::shared_ptr<arrow::DataType>& element_type) {
-    PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
-                           CoreOptions::FromMap({{"fields.f.distinct", "true"}}));
-    return FieldCollectAgg::Create(arrow::list(element_type), options, "f", GetDefaultPool());
-}
-
-VariantType Array(std::vector<VariantType> values) {
-    return VariantType(
-        checked_pointer_cast<InternalArray>(std::make_shared<GenericArray>(std::move(values))));
-}
-
 VariantType IntStringRow(int32_t id, std::string_view name) {
     std::shared_ptr<GenericRow> row = std::make_shared<GenericRow>(2);
     row->SetField(0, id);
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg.cpp b/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg.cpp
index 2cd48babf..107d46a0f 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg.cpp
@@ -19,6 +19,7 @@
 
 #include "paimon/core/mergetree/compact/aggregate/field_merge_map_agg.h"
 
+#include <unordered_map>
 #include <utility>
 #include <vector>
 
@@ -33,6 +34,10 @@
 namespace paimon {
 namespace {
 
+constexpr int32_t kHashThreshold = 192;
+using detail::SemanticEqual;
+using detail::SemanticHash;
+
 struct MapEntry {
     VariantType key;
     VariantType value;
@@ -69,6 +74,29 @@ Status PutMap(const std::shared_ptr<InternalMap>& map,
     return Status::OK();
 }
 
+Status PutMapWithHash(
+    const std::shared_ptr<InternalMap>& map, const std::shared_ptr<arrow::DataType>& key_type,
+    const std::shared_ptr<arrow::DataType>& value_type,
+    std::unordered_map<VariantType, int32_t, SemanticHash, SemanticEqual>* key_index,
+    std::vector<MapEntry>* entries) {
+    std::shared_ptr<InternalArray> keys = map->KeyArray();
+    std::shared_ptr<InternalArray> values = map->ValueArray();
+    for (int32_t i = 0; i < map->Size(); ++i) {
+        PAIMON_ASSIGN_OR_RAISE(VariantType key, FieldAggregateUtils::GetValue(*keys, i, key_type));
+        PAIMON_ASSIGN_OR_RAISE(VariantType value,
+                               FieldAggregateUtils::GetValue(*values, i, value_type));
+        auto iter = key_index->find(key);
+        if (iter != key_index->end()) {
+            entries->at(iter->second).value = std::move(value);
+        } else {
+            auto new_index = static_cast<int32_t>(entries->size());
+            entries->push_back(MapEntry{std::move(key), std::move(value)});
+            key_index->emplace(entries->back().key, new_index);
+        }
+    }
+    return Status::OK();
+}
+
 VariantType MakeMap(std::vector<MapEntry> entries,
                     std::vector<std::shared_ptr<InternalArray>> key_holders,
                     std::vector<std::shared_ptr<InternalArray>> value_holders) {
@@ -119,8 +147,21 @@ Result<VariantType> FieldMergeMapAgg::AggImpl(const VariantType& accumulator,
     auto input_map = DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(input_field);
     std::vector<MapEntry> entries;
     entries.reserve(accumulator_map->Size() + input_map->Size());
-    PAIMON_RETURN_NOT_OK(PutMap(accumulator_map, key_type_, value_type_, &entries));
-    PAIMON_RETURN_NOT_OK(PutMap(input_map, key_type_, value_type_, &entries));
+    int32_t total_size = accumulator_map->Size() + input_map->Size();
+    if (total_size >= kHashThreshold && FieldAggregateUtils::IsHashableType(key_type_)) {
+        SemanticHash hasher{key_type_};
+        SemanticEqual equal{key_type_};
+        std::unordered_map<VariantType, int32_t, SemanticHash, SemanticEqual> key_index(0, hasher,
+                                                                                        equal);
+        key_index.reserve(total_size);
+        PAIMON_RETURN_NOT_OK(
+            PutMapWithHash(accumulator_map, key_type_, value_type_, &key_index, &entries));
+        PAIMON_RETURN_NOT_OK(
+            PutMapWithHash(input_map, key_type_, value_type_, &key_index, &entries));
+    } else {
+        PAIMON_RETURN_NOT_OK(PutMap(accumulator_map, key_type_, value_type_, &entries));
+        PAIMON_RETURN_NOT_OK(PutMap(input_map, key_type_, value_type_, &entries));
+    }
     return MakeMap(std::move(entries), {accumulator_map->KeyArray(), input_map->KeyArray()},
                    {accumulator_map->ValueArray(), input_map->ValueArray()});
 }
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg_test.cpp b/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg_test.cpp
index 1dcbd3041..e867623bb 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg_test.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_merge_map_agg_test.cpp
@@ -20,6 +20,7 @@
 #include "paimon/core/mergetree/compact/aggregate/field_merge_map_agg.h"
 
 #include <memory>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -52,6 +53,15 @@ int32_t FindValue(const VariantType& value, int32_t key) {
     return -1;
 }
 
+std::vector<int32_t> Keys(const VariantType& value) {
+    auto map = DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(value);
+    std::vector<int32_t> keys;
+    for (int32_t i = 0; i < map->Size(); ++i) {
+        keys.push_back(map->KeyArray()->GetInt(i));
+    }
+    return keys;
+}
+
 }  // namespace
 
 TEST(FieldMergeMapAggTest, InputOverwritesAndRetractUsesKeys) {
@@ -80,6 +90,72 @@ TEST(FieldMergeMapAggTest, InputOverwritesAndRetractUsesKeys) {
     ASSERT_EQ(-1, FindValue(retracted, 2));
 }
 
+TEST(FieldMergeMapAggTest, HashPathDeduplicatesAccumulatorAndPreservesOrder) {
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldMergeMapAgg> agg,
+                         FieldMergeMapAgg::Create(arrow::map(arrow::int32(), arrow::int32()), "f",
+                                                  GetDefaultPool()));
+    std::vector<VariantType> accumulator_keys;
+    std::vector<VariantType> accumulator_values;
+    for (int32_t i = 0; i <= 100; ++i) {
+        accumulator_keys.emplace_back(i);
+        accumulator_values.emplace_back(100 + i);
+    }
+    accumulator_keys.emplace_back(10);
+    accumulator_values.emplace_back(1010);
+
+    std::vector<VariantType> input_keys;
+    std::vector<VariantType> input_values;
+    for (int32_t i = 50; i <= 140; ++i) {
+        input_keys.emplace_back(i);
+        input_values.emplace_back(2000 + i);
+    }
+    input_keys.emplace_back(60);
+    input_values.emplace_back(2060);
+
+    ASSERT_OK_AND_ASSIGN(
+        VariantType result,
+        agg->Agg(IntMap(std::move(accumulator_keys), std::move(accumulator_values)),
+                 IntMap(std::move(input_keys), std::move(input_values))));
+    std::vector<int32_t> expected_keys;
+    for (int32_t i = 0; i <= 140; ++i) {
+        expected_keys.push_back(i);
+    }
+    ASSERT_EQ(expected_keys, Keys(result));
+    ASSERT_EQ(1010, FindValue(result, 10));
+    ASSERT_EQ(2050, FindValue(result, 50));
+    ASSERT_EQ(2060, FindValue(result, 60));
+    ASSERT_EQ(2140, FindValue(result, 140));
+}
+
+TEST(FieldMergeMapAggTest, HashPathUsesStringContentAndOverwrites) {
+    std::vector<std::shared_ptr<arrow::DataType>> key_types{arrow::utf8(), arrow::binary()};
+    for (const auto& key_type : key_types) {
+        ASSERT_OK_AND_ASSIGN(
+            std::unique_ptr<FieldMergeMapAgg> agg,
+            FieldMergeMapAgg::Create(arrow::map(key_type, arrow::int32()), "f", GetDefaultPool()));
+        std::vector<std::string> left(100, "same-content-with-a-longer-buffer");
+        std::vector<std::string> right(100, "same-content-with-a-longer-buffer");
+        std::vector<VariantType> left_keys;
+        std::vector<VariantType> right_keys;
+        std::vector<VariantType> left_values(100, VariantType(int32_t{10}));
+        std::vector<VariantType> right_values(100, VariantType(int32_t{20}));
+        for (const auto& value : left) {
+            left_keys.emplace_back(std::string_view(value));
+        }
+        for (const auto& value : right) {
+            right_keys.emplace_back(std::string_view(value));
+        }
+
+        ASSERT_OK_AND_ASSIGN(VariantType result,
+                             agg->Agg(IntMap(std::move(left_keys), std::move(left_values)),
+                                      IntMap(std::move(right_keys), std::move(right_values))));
+        auto result_map = DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(result);
+        ASSERT_EQ(1, result_map->Size());
+        ASSERT_EQ("same-content-with-a-longer-buffer", result_map->KeyArray()->GetStringView(0));
+        ASSERT_EQ(20, result_map->ValueArray()->GetInt(0));
+    }
+}
+
 TEST(FieldMergeMapAggTest, NullAndTypeValidation) {
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldMergeMapAgg> agg,
                          FieldMergeMapAgg::Create(arrow::map(arrow::int32(), arrow::int32()), "f",
@@ -87,6 +163,11 @@ TEST(FieldMergeMapAggTest, NullAndTypeValidation) {
     VariantType map = IntMap({int32_t{1}}, {int32_t{10}});
     ASSERT_OK_AND_ASSIGN(VariantType result, agg->Agg(VariantType(NullType()), map));
     ASSERT_EQ(10, FindValue(result, 1));
+    ASSERT_OK_AND_ASSIGN(VariantType accumulator_result, agg->Agg(map, VariantType(NullType())));
+    ASSERT_EQ(10, FindValue(accumulator_result, 1));
+    ASSERT_OK_AND_ASSIGN(VariantType null_result,
+                         agg->Agg(VariantType(NullType()), VariantType(NullType())));
+    ASSERT_TRUE(DataDefine::IsVariantNull(null_result));
     ASSERT_NOK(FieldMergeMapAgg::Create(arrow::int32(), "f", GetDefaultPool()));
 }