From 17dd53fd88752485a1436b8cbdc73e3d6a84b3c8 Mon Sep 17 00:00:00 2001 From: VietCT04 Date: Tue, 15 Sep 2026 17:42:16 +0800 Subject: [PATCH 1/4] perf(aggregate): use hash lookup for scalar field aggregators --- .../aggregate/field_aggregate_utils.cpp | 87 +++++++++ .../compact/aggregate/field_aggregate_utils.h | 33 ++++ .../compact/aggregate/field_collect_agg.cpp | 46 ++++- .../aggregate/field_collect_agg_test.cpp | 171 ++++++++++++++++++ .../compact/aggregate/field_merge_map_agg.cpp | 45 ++++- .../aggregate/field_merge_map_agg_test.cpp | 81 +++++++++ 6 files changed, 454 insertions(+), 9 deletions(-) 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..acd907496 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp +++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp @@ -19,8 +19,13 @@ #include "paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h" +#include +#include +#include #include +#include #include +#include #include #include @@ -123,6 +128,25 @@ Result EqualMaps(const std::shared_ptr& lhs, return true; } +size_t CombineHash(size_t lhs, size_t rhs) { + return lhs ^ (rhs + static_cast(0x9e3779b97f4a7c15ULL) + (lhs << 6) + (lhs >> 2)); +} + +template +size_t HashFloatingPoint(T value) { + using Bits = std::conditional_t; + Bits bits; + std::memcpy(&bits, &value, sizeof(value)); + if (std::isnan(value)) { + if constexpr (sizeof(T) == sizeof(uint32_t)) { + bits = static_cast(0x7fc00000U); + } else { + bits = static_cast(0x7ff8000000000000ULL); + } + } + return std::hash{}(bits); +} + } // namespace VariantType FieldAggregateUtils::OwnedBinary(const VariantType& value, MemoryPool* pool) { @@ -244,4 +268,67 @@ Result FieldAggregateUtils::Equals(const VariantType& lhs, const VariantTy } } +bool FieldAggregateUtils::IsHashableType(const std::shared_ptr& 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& type) { + assert(IsHashableType(type)); + size_t result = std::hash{}(static_cast(type->id())); + if (DataDefine::IsVariantNull(value)) { + return result; + } + switch (type->id()) { + case arrow::Type::BOOL: + return CombineHash(result, std::hash{}(DataDefine::GetVariantValue(value))); + case arrow::Type::INT8: + return CombineHash(result, std::hash{}(DataDefine::GetVariantValue(value))); + case arrow::Type::INT16: + return CombineHash( + result, std::hash{}(DataDefine::GetVariantValue(value))); + case arrow::Type::INT32: + case arrow::Type::DATE32: + return CombineHash( + result, std::hash{}(DataDefine::GetVariantValue(value))); + case arrow::Type::INT64: + return CombineHash( + result, std::hash{}(DataDefine::GetVariantValue(value))); + case arrow::Type::FLOAT: + return CombineHash(result, + HashFloatingPoint(DataDefine::GetVariantValue(value))); + case arrow::Type::DOUBLE: + return CombineHash(result, + HashFloatingPoint(DataDefine::GetVariantValue(value))); + case arrow::Type::STRING: + case arrow::Type::BINARY: + return CombineHash(result, + std::hash{}(DataDefine::GetStringView(value))); + case arrow::Type::TIMESTAMP: { + const auto& timestamp = DataDefine::GetVariantValue(value); + return CombineHash(result, + CombineHash(std::hash{}(timestamp.GetMillisecond()), + std::hash{}(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..0571cf3f4 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 +#include #include #include "arrow/type_fwd.h" @@ -62,6 +64,37 @@ class FieldAggregateUtils { /// @return Whether the values are equal, or an error Status. static Result Equals(const VariantType& lhs, const VariantType& rhs, const std::shared_ptr& type); + + /// Return whether values of this logical type have a hash compatible with Equals(). + static bool IsHashableType(const std::shared_ptr& 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& type); }; +namespace detail { + +struct SemanticHash { + std::shared_ptr type; + + size_t operator()(const VariantType& value) const { + return FieldAggregateUtils::Hash(value, type); + } +}; + +struct SemanticEqual { + std::shared_ptr type; + + bool operator()(const VariantType& lhs, const VariantType& rhs) const { + Result 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..aaa3c9b99 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 #include #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 Contains(const std::vector& values, const VariantType& candidate, const std::shared_ptr& element_type) { for (const VariantType& value : values) { @@ -68,6 +72,23 @@ Status AppendArray(const std::shared_ptr& array, return Status::OK(); } +Status AppendArrayWithHash(const std::shared_ptr& array, + const std::shared_ptr& element_type, + std::unordered_set* seen, + std::vector* 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> FieldCollectAgg::Create( @@ -111,12 +132,23 @@ Result FieldCollectAgg::AggImpl(const VariantType& accumulator, std::shared_ptr input_array = input_null ? nullptr : DataDefine::GetVariantValue>(input_field); + int32_t total_size = (accumulator_array ? accumulator_array->Size() : 0) + + (input_array ? input_array->Size() : 0); std::vector 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 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> 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..cd26b9fee 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 +#include +#include #include #include #include @@ -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,11 @@ Result> MakeCollectAgg(bool distinct) { return FieldCollectAgg::Create(arrow::list(arrow::int32()), options, "f", GetDefaultPool()); } +Result> MakeDistinctAgg( + const std::shared_ptr& element_type); + +VariantType Array(std::vector values); + } // namespace TEST(FieldCollectAggTest, ConcatenatesWithoutReversing) { @@ -94,6 +102,169 @@ TEST(FieldCollectAggTest, DistinctAndRetractOneOccurrence) { ASSERT_EQ((std::vector{1, 2, 3}), Values(retract_result)); } +TEST(FieldCollectAggTest, HashPathPreservesDuplicatesOverlapAndOrder) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, MakeCollectAgg(true)); + std::vector accumulator; + std::vector 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 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 int_agg, MakeCollectAgg(true)); + std::vector accumulator; + std::vector 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>(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 string_agg, + MakeDistinctAgg(element_type)); + std::vector left(100, "same-content"); + std::vector right(100, "same-content"); + std::vector left_values; + std::vector 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>(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& type, + const VariantType& negative_zero, const VariantType& nan1, + const VariantType& positive_zero, const VariantType& nan2) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, + MakeDistinctAgg(type)); + std::vector accumulator; + std::vector 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>(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 timestamp_agg, + MakeDistinctAgg(arrow::timestamp(arrow::TimeUnit::NANO))); + Timestamp first = Timestamp::FromEpochMillis(100, 7); + Timestamp second = Timestamp::FromEpochMillis(101, 0); + std::vector timestamps; + std::vector 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>(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 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 accumulator; + std::vector 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>(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 agg, + MakeDistinctAgg(arrow::list(arrow::int32()))); + std::vector accumulator; + std::vector 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>(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())); 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..10b19369a 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 #include #include @@ -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& map, return Status::OK(); } +Status PutMapWithHash( + const std::shared_ptr& map, const std::shared_ptr& key_type, + const std::shared_ptr& value_type, + std::unordered_map* key_index, + std::vector* entries) { + std::shared_ptr keys = map->KeyArray(); + std::shared_ptr 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 { + int32_t new_index = static_cast(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 entries, std::vector> key_holders, std::vector> value_holders) { @@ -119,8 +147,21 @@ Result FieldMergeMapAgg::AggImpl(const VariantType& accumulator, auto input_map = DataDefine::GetVariantValue>(input_field); std::vector 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 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..4e9541aa8 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 +#include #include #include @@ -52,6 +53,15 @@ int32_t FindValue(const VariantType& value, int32_t key) { return -1; } +std::vector Keys(const VariantType& value) { + auto map = DataDefine::GetVariantValue>(value); + std::vector 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 agg, + FieldMergeMapAgg::Create(arrow::map(arrow::int32(), arrow::int32()), "f", + GetDefaultPool())); + std::vector accumulator_keys; + std::vector 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 input_keys; + std::vector 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 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> key_types{arrow::utf8(), arrow::binary()}; + for (const auto& key_type : key_types) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, + FieldMergeMapAgg::Create(arrow::map(key_type, arrow::int32()), "f", + GetDefaultPool())); + std::vector left(100, "same-content-with-a-longer-buffer"); + std::vector right(100, "same-content-with-a-longer-buffer"); + std::vector left_keys; + std::vector right_keys; + std::vector left_values(100, VariantType(int32_t{10})); + std::vector 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>(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 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())); } From 618e749d3f5ba79651b89c75d2364dd3e7df1442 Mon Sep 17 00:00:00 2001 From: VietCT04 Date: Tue, 15 Sep 2026 18:06:15 +0800 Subject: [PATCH 2/4] fix clang format --- .../aggregate/field_aggregate_utils.cpp | 14 +++++------ .../compact/aggregate/field_aggregate_utils.h | 3 +-- .../compact/aggregate/field_collect_agg.cpp | 3 +-- .../aggregate/field_collect_agg_test.cpp | 25 ++++++++----------- .../compact/aggregate/field_merge_map_agg.cpp | 4 +-- .../aggregate/field_merge_map_agg_test.cpp | 14 +++++------ 6 files changed, 29 insertions(+), 34 deletions(-) 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 acd907496..c9236cab9 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp +++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp @@ -288,7 +288,7 @@ bool FieldAggregateUtils::IsHashableType(const std::shared_ptr& } size_t FieldAggregateUtils::Hash(const VariantType& value, - const std::shared_ptr& type) { + const std::shared_ptr& type) { assert(IsHashableType(type)); size_t result = std::hash{}(static_cast(type->id())); if (DataDefine::IsVariantNull(value)) { @@ -300,15 +300,15 @@ size_t FieldAggregateUtils::Hash(const VariantType& value, case arrow::Type::INT8: return CombineHash(result, std::hash{}(DataDefine::GetVariantValue(value))); case arrow::Type::INT16: - return CombineHash( - result, std::hash{}(DataDefine::GetVariantValue(value))); + return CombineHash(result, + std::hash{}(DataDefine::GetVariantValue(value))); case arrow::Type::INT32: case arrow::Type::DATE32: - return CombineHash( - result, std::hash{}(DataDefine::GetVariantValue(value))); + return CombineHash(result, + std::hash{}(DataDefine::GetVariantValue(value))); case arrow::Type::INT64: - return CombineHash( - result, std::hash{}(DataDefine::GetVariantValue(value))); + return CombineHash(result, + std::hash{}(DataDefine::GetVariantValue(value))); case arrow::Type::FLOAT: return CombineHash(result, HashFloatingPoint(DataDefine::GetVariantValue(value))); 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 0571cf3f4..db03de558 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h +++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h @@ -71,8 +71,7 @@ class FieldAggregateUtils { /// 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& type); + static size_t Hash(const VariantType& value, const std::shared_ptr& type); }; namespace detail { 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 aaa3c9b99..919286b70 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp +++ b/src/paimon/core/mergetree/compact/aggregate/field_collect_agg.cpp @@ -142,8 +142,7 @@ Result FieldCollectAgg::AggImpl(const VariantType& accumulator, SemanticEqual equal{element_type_}; std::unordered_set seen(0, hasher, equal); seen.reserve(total_size); - PAIMON_RETURN_NOT_OK( - AppendArrayWithHash(accumulator_array, element_type_, &seen, &values)); + 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)); 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 cd26b9fee..7b87e43e2 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 @@ -128,9 +128,8 @@ TEST(FieldCollectAggTest, HashPathHandlesNullsAndStringContent) { 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)))); + ASSERT_OK_AND_ASSIGN(VariantType int_result, int_agg->Agg(IntArray(std::move(accumulator)), + IntArray(std::move(input)))); auto int_array = DataDefine::GetVariantValue>(int_result); ASSERT_EQ(3, int_array->Size()); ASSERT_TRUE(int_array->IsNullAt(0)); @@ -150,9 +149,9 @@ TEST(FieldCollectAggTest, HashPathHandlesNullsAndStringContent) { 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)))); + 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>(string_result); ASSERT_EQ(1, string_array->Size()); @@ -177,17 +176,15 @@ TEST(FieldCollectAggTest, HashPathPreservesFloatingPointAndTimestampSemantics) { auto verify_floating_point = [&](const std::shared_ptr& type, const VariantType& negative_zero, const VariantType& nan1, const VariantType& positive_zero, const VariantType& nan2) { - ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, - MakeDistinctAgg(type)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, MakeDistinctAgg(type)); std::vector accumulator; std::vector 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)))); + ASSERT_OK_AND_ASSIGN(VariantType result, agg->Agg(IntArray(std::move(accumulator)), + IntArray(std::move(input)))); auto array = DataDefine::GetVariantValue>(result); ASSERT_EQ(3, array->Size()); if (type->id() == arrow::Type::FLOAT) { @@ -215,9 +212,9 @@ TEST(FieldCollectAggTest, HashPathPreservesFloatingPointAndTimestampSemantics) { 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)))); + ASSERT_OK_AND_ASSIGN( + VariantType timestamp_result, + timestamp_agg->Agg(IntArray(std::move(timestamps)), IntArray(std::move(more_timestamps)))); auto timestamp_array = DataDefine::GetVariantValue>(timestamp_result); ASSERT_EQ(2, timestamp_array->Size()); 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 10b19369a..da579de70 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 @@ -151,8 +151,8 @@ Result FieldMergeMapAgg::AggImpl(const VariantType& accumulator, if (total_size >= kHashThreshold && FieldAggregateUtils::IsHashableType(key_type_)) { SemanticHash hasher{key_type_}; SemanticEqual equal{key_type_}; - std::unordered_map key_index( - 0, hasher, equal); + std::unordered_map key_index(0, hasher, + equal); key_index.reserve(total_size); PAIMON_RETURN_NOT_OK( PutMapWithHash(accumulator_map, key_type_, value_type_, &key_index, &entries)); 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 4e9541aa8..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 @@ -112,10 +112,10 @@ TEST(FieldMergeMapAggTest, HashPathDeduplicatesAccumulatorAndPreservesOrder) { 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)))); + 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 expected_keys; for (int32_t i = 0; i <= 140; ++i) { expected_keys.push_back(i); @@ -130,9 +130,9 @@ TEST(FieldMergeMapAggTest, HashPathDeduplicatesAccumulatorAndPreservesOrder) { TEST(FieldMergeMapAggTest, HashPathUsesStringContentAndOverwrites) { std::vector> key_types{arrow::utf8(), arrow::binary()}; for (const auto& key_type : key_types) { - ASSERT_OK_AND_ASSIGN(std::unique_ptr agg, - FieldMergeMapAgg::Create(arrow::map(key_type, arrow::int32()), "f", - GetDefaultPool())); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr agg, + FieldMergeMapAgg::Create(arrow::map(key_type, arrow::int32()), "f", GetDefaultPool())); std::vector left(100, "same-content-with-a-longer-buffer"); std::vector right(100, "same-content-with-a-longer-buffer"); std::vector left_keys; From 78076fa0b8697552546d04890040699b8d024151 Mon Sep 17 00:00:00 2001 From: VietCT04 Date: Tue, 15 Sep 2026 22:19:41 +0800 Subject: [PATCH 3/4] refactor(aggregate): reuse floating-point canonicalization utilities --- .../aggregate/field_aggregate_utils.cpp | 26 ++++--------------- 1 file changed, 5 insertions(+), 21 deletions(-) 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 c9236cab9..f9a98f1bb 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp +++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp @@ -20,12 +20,10 @@ #include "paimon/core/mergetree/compact/aggregate/field_aggregate_utils.h" #include -#include #include #include #include #include -#include #include #include @@ -39,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" @@ -132,21 +131,6 @@ size_t CombineHash(size_t lhs, size_t rhs) { return lhs ^ (rhs + static_cast(0x9e3779b97f4a7c15ULL) + (lhs << 6) + (lhs >> 2)); } -template -size_t HashFloatingPoint(T value) { - using Bits = std::conditional_t; - Bits bits; - std::memcpy(&bits, &value, sizeof(value)); - if (std::isnan(value)) { - if constexpr (sizeof(T) == sizeof(uint32_t)) { - bits = static_cast(0x7fc00000U); - } else { - bits = static_cast(0x7ff8000000000000ULL); - } - } - return std::hash{}(bits); -} - } // namespace VariantType FieldAggregateUtils::OwnedBinary(const VariantType& value, MemoryPool* pool) { @@ -310,11 +294,11 @@ size_t FieldAggregateUtils::Hash(const VariantType& value, return CombineHash(result, std::hash{}(DataDefine::GetVariantValue(value))); case arrow::Type::FLOAT: - return CombineHash(result, - HashFloatingPoint(DataDefine::GetVariantValue(value))); + return CombineHash(result, std::hash{}(CanonicalizeFloatToIntBits( + DataDefine::GetVariantValue(value)))); case arrow::Type::DOUBLE: - return CombineHash(result, - HashFloatingPoint(DataDefine::GetVariantValue(value))); + return CombineHash(result, std::hash{}(CanonicalizeDoubleToLongBits( + DataDefine::GetVariantValue(value)))); case arrow::Type::STRING: case arrow::Type::BINARY: return CombineHash(result, From 75f9ad37cafce8b11eab0cad17b91ad7b891c718 Mon Sep 17 00:00:00 2001 From: VietCT04 Date: Wed, 16 Sep 2026 23:42:00 +0800 Subject: [PATCH 4/4] modified base on review --- .../aggregate/field_aggregate_utils.cpp | 2 +- .../aggregate/field_collect_agg_test.cpp | 23 ++++++++----------- .../compact/aggregate/field_merge_map_agg.cpp | 2 +- 3 files changed, 11 insertions(+), 16 deletions(-) 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 f9a98f1bb..8dc4b49ff 100644 --- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp +++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp @@ -274,7 +274,7 @@ bool FieldAggregateUtils::IsHashableType(const std::shared_ptr& size_t FieldAggregateUtils::Hash(const VariantType& value, const std::shared_ptr& type) { assert(IsHashableType(type)); - size_t result = std::hash{}(static_cast(type->id())); + size_t result = std::hash{}(static_cast(type->id())); if (DataDefine::IsVariantNull(value)) { return result; } 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 7b87e43e2..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 @@ -64,9 +64,16 @@ Result> MakeCollectAgg(bool distinct) { } Result> MakeDistinctAgg( - const std::shared_ptr& element_type); + const std::shared_ptr& 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 values); +VariantType Array(std::vector values) { + return VariantType( + checked_pointer_cast(std::make_shared(std::move(values)))); +} } // namespace @@ -271,18 +278,6 @@ TEST(FieldCollectAggTest, RejectsNonArrayType) { // collection over composite element types. namespace { -Result> MakeDistinctAgg( - const std::shared_ptr& 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 values) { - return VariantType( - checked_pointer_cast(std::make_shared(std::move(values)))); -} - VariantType IntStringRow(int32_t id, std::string_view name) { std::shared_ptr row = std::make_shared(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 da579de70..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 @@ -89,7 +89,7 @@ Status PutMapWithHash( if (iter != key_index->end()) { entries->at(iter->second).value = std::move(value); } else { - int32_t new_index = static_cast(entries->size()); + auto new_index = static_cast(entries->size()); entries->push_back(MapEntry{std::move(key), std::move(value)}); key_index->emplace(entries->back().key, new_index); }