This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 292e6e01 perf(aggregate): use hash lookup for scalar field aggregators
(#349)
292e6e01 is described below
commit 292e6e016fca81a40dd60cb1cae93443fd627fb4
Author: VietCT04 <[email protected]>
AuthorDate: Thu Sep 17 11:18:46 2026 +0800
perf(aggregate): use hash lookup for scalar field aggregators (#349)
---
.../compact/aggregate/field_aggregate_utils.cpp | 71 ++++++++
.../compact/aggregate/field_aggregate_utils.h | 32 ++++
.../compact/aggregate/field_collect_agg.cpp | 45 ++++-
.../compact/aggregate/field_collect_agg_test.cpp | 187 +++++++++++++++++++--
.../compact/aggregate/field_merge_map_agg.cpp | 45 ++++-
.../compact/aggregate/field_merge_map_agg_test.cpp | 81 +++++++++
6 files changed, 440 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 22e02f10..8dc4b49f 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 e3f74ed7..db03de55 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 ff938b73..919286b7 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 33804567..22389696 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 2cd48bab..107d46a0 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 1dcbd304..e867623b 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()));
}
