From f4dd7a03c1868330f97deff3e218d687fbc4ea79 Mon Sep 17 00:00:00 2001
From: nooneuse <nooneuse@users.noreply.github.com>
Date: Sat, 30 May 2026 18:36:41 +0800
Subject: [PATCH] Add datasketches HLL sketch aggregate functions (#63143)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

### What problem does this PR solve?

> An aggregate function is required to process user data containing
Datasketches HLL sketches. In many data aggregation scenarios, users
pre‑aggregate detailed data in Hive using the sketching techniques
provided by Apache Datasketches, and then analyze the resulting sketches
across various OLAP engines. Compared with the HLL union aggregate
functions natively offered by these engines, there are two key diff to
using Datasketches HLL sketches: firstly, the use cases differ; and
secondly, HLL sketches can be used seamlessly across different
engines—for example, simultaneously in ES, Doris, and ClickHouse. Such
requirements are common in many production environments.

Issue Number:
- #63142(https://github.com/apache/doris/issues/63142)
- #26416
- #56246

Summary:
Implemented a built-in aggregate function that integrates the
Datasketches HLL sketch. This aggregate function cannot rely on the Java
UDF environment. Considering that in the Java UDF environment, Strings
are encoded in UTF-8, which corrupts the binary data of sketches, the
serialization/deserialization operations for sketches must be
implemented on the BE side. (additionally, since Apache Datasketches has
been added to the contrib directory via a git submodule, it will become
very easy to add other sketches such as theta sketch in the future.)

**see**: https://github.com/apache/doris/issues/63142
**use case**: see regression test &
https://github.com/apache/doris/issues/63142

---------

Co-authored-by: yuanyuhao <yuanyuhao@bytedance.com>
---
 .gitmodules                                   |    3 +
 ...te_function_datasketches_hll_union_agg.cpp |   44 +
 ...gate_function_datasketches_hll_union_agg.h |  243 ++++
 .../aggregate_function_simple_factory.cpp     |    3 +
 .../agg_datasketches_hll_union_agg_test.cpp   | 1097 +++++++++++++++++
 build.sh                                      |    9 +
 contrib/datasketches-cpp                      |    1 +
 .../catalog/BuiltinAggregateFunctions.java    |    3 +
 .../agg/DataSketchesHllUnionAgg.java          |  113 ++
 .../visitor/AggregateFunctionVisitor.java     |    5 +
 .../test_datasketches_hll_union_agg.out       |   28 +
 .../test_datasketches_hll_union_agg.groovy    |  170 +++
 run-be-ut.sh                                  |   14 +-
 13 files changed, 1732 insertions(+), 1 deletion(-)
 create mode 100644 be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.cpp
 create mode 100644 be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
 create mode 100644 be/test/exprs/aggregate/agg_datasketches_hll_union_agg_test.cpp
 create mode 160000 contrib/datasketches-cpp
 create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/DataSketchesHllUnionAgg.java
 create mode 100644 regression-test/data/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.out
 create mode 100644 regression-test/suites/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.groovy

diff --git a/.gitmodules b/.gitmodules
index 54c1a8a36366af..eb8e703aa8af98 100644
--- a/.gitmodules
+++ b/.gitmodules
@@ -29,3 +29,6 @@
 	path = contrib/openblas
 	url = https://github.com/apache/doris-thirdparty.git
 	branch = openblas
+[submodule "contrib/datasketches-cpp"]
+	path = contrib/datasketches-cpp
+	url = https://github.com/apache/datasketches-cpp.git
diff --git a/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.cpp b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.cpp
new file mode 100644
index 00000000000000..c9b7013e7a960f
--- /dev/null
+++ b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.cpp
@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h"
+
+#include <string>
+
+#include "core/data_type/data_type.h"
+#include "core/data_type/define_primitive_type.h"
+#include "exec/common/hash_table/hash.h" // IWYU pragma: keep
+#include "exprs/aggregate/aggregate_function_simple_factory.h"
+#include "exprs/aggregate/helpers.h"
+namespace doris {
+template <template <PrimitiveType> class Data>
+AggregateFunctionPtr create_aggregate_function_datasketches_hll_union_agg(
+        const std::string& name, const DataTypes& argument_types, const DataTypePtr& result_type,
+        const bool result_is_nullable, const AggregateFunctionAttr& attr) {
+    return creator_with_type_list<TYPE_STRING, TYPE_VARCHAR, TYPE_VARBINARY>::create<
+            AggregateFunctionDataSketchesHllUnionAgg, Data>(argument_types, result_is_nullable,
+                                                            attr);
+}
+void register_aggregate_function_datasketches_HLL_union_agg(
+        AggregateFunctionSimpleFactory& factory) {
+    AggregateFunctionCreator creator =
+            create_aggregate_function_datasketches_hll_union_agg<AggregateFunctionHllSketchData>;
+    factory.register_function_both("datasketches_hll_union_agg", creator);
+    factory.register_alias("datasketches_hll_union_agg", "ds_hll_estimate");
+    factory.register_alias("datasketches_hll_union_agg", "datasketches_hll_estimate");
+}
+} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
new file mode 100644
index 00000000000000..d9f82f193e82c5
--- /dev/null
+++ b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
@@ -0,0 +1,243 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+#include <stddef.h>
+
+#include <DataSketches/hll.hpp>
+#include <algorithm>
+#include <boost/iterator/iterator_facade.hpp>
+#include <memory>
+#include <optional>
+#include <type_traits>
+#include <vector>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "core/assert_cast.h"
+#include "core/column/column.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/define_primitive_type.h"
+#include "core/field.h"
+#include "core/string_ref.h"
+#include "core/types.h"
+#include "core/uint128.h"
+#include "exec/common/hash_table/hash.h"
+#include "exec/common/hash_table/phmap_fwd_decl.h"
+#include "exprs/aggregate/aggregate_function.h"
+#include "util/var_int.h"
+template <typename T>
+struct HashCRC32;
+namespace doris {
+class Arena;
+class BufferReadable;
+class BufferWritable;
+template <PrimitiveType T>
+class ColumnDecimal;
+/// datasketches_hll_union_agg
+template <PrimitiveType T>
+struct AggregateFunctionHllSketchData {
+    /** We set the default LgK to 12,
+      * as this value is used as a performance baseline in the relevant documentation.
+      * (https://datasketches.apache.org/docs/HLL/HllPerformance.html)
+      */
+    static constexpr uint8_t DEFAULT_LOG_K = 12;
+    using Alloc = CustomStdAllocator<uint8_t>;
+    using Sketch = datasketches::hll_sketch_alloc<Alloc>;
+    using Union = datasketches::hll_union_alloc<Alloc>;
+
+    std::optional<Union> hll_union_data;
+
+    static String get_name() { return "datasketches_hll_union_agg"; }
+
+    void merge(const Sketch& sketch_data) {
+        if (!hll_union_data.has_value()) {
+            /** We clamp max lg_k to [7, 21],
+              * considering that the code comment requires 7 to 21.
+              * See: datasketches-cpp/hll/include/hll.hpp:451
+              */
+            constexpr uint8_t MIN_UNION_LOG_K = 7;
+            const uint8_t union_lg_k =
+                    std::clamp<uint8_t>(sketch_data.get_lg_config_k(), MIN_UNION_LOG_K,
+                                        datasketches::hll_constants::MAX_LOG_K);
+            hll_union_data.emplace(union_lg_k, Alloc());
+        }
+        try {
+            hll_union_data->update(sketch_data);
+        } catch (const doris::Exception& e) {
+            throw Exception(e.code(), "Internal error happened when update HLL sketch: {}",
+                            e.to_string());
+        } catch (const std::exception& e) {
+            throw Exception(ErrorCode::INTERNAL_ERROR,
+                            "Internal error happened when update HLL sketch: {}", e.what());
+        } catch (...) {
+            throw Exception(ErrorCode::INTERNAL_ERROR,
+                            "Internal error happened when update HLL sketch: unknown exception.");
+        }
+    }
+    void reset() {
+        if (hll_union_data.has_value()) {
+            hll_union_data->reset();
+        }
+        hll_union_data.reset();
+    }
+
+    void write_sketch(BufferWritable& buf, const Sketch& sk) const {
+        auto serialized_bytes = sk.serialize_compact();
+        StringRef d(serialized_bytes.data(), serialized_bytes.size());
+        buf.write_binary(d);
+    }
+    void write(BufferWritable& buf) const {
+        if (!hll_union_data.has_value()) {
+            /** Using DEFAULT_LOG_K(12) here is surely sufficient,
+              * because in this case the union that actually needs to be serialized should contain no data.
+              */
+            Union u(DEFAULT_LOG_K, Alloc());
+            write_sketch(buf, u.get_result());
+            return;
+        }
+        try {
+            auto cache = hll_union_data->get_result();
+            write_sketch(buf, cache);
+        } catch (const doris::Exception& e) {
+            throw Exception(e.code(), "Internal error happened when serialize HLL sketch: {}",
+                            e.to_string());
+        } catch (const std::exception& e) {
+            throw Exception(ErrorCode::INTERNAL_ERROR,
+                            "Internal error happened when serialize HLL sketch: {}", e.what());
+        } catch (...) {
+            throw Exception(
+                    ErrorCode::INTERNAL_ERROR,
+                    "Internal error happened when serialize HLL sketch: unknown exception.");
+        }
+    }
+    void read(BufferReadable& buf) {
+        StringRef d;
+        buf.read_binary(d);
+
+        auto cache = [&]() -> Sketch {
+            try {
+                return Sketch::deserialize(d.data, d.size, Alloc());
+            } catch (const doris::Exception& e) {
+                throw Exception(e.code(), "Failed to deserialize HLL sketch when read: {}",
+                                e.to_string());
+            } catch (const std::exception& e) {
+                throw Exception(ErrorCode::CORRUPTION, "HLL sketch data corrupted when read: {}",
+                                e.what());
+            } catch (...) {
+                throw Exception(ErrorCode::CORRUPTION,
+                                "HLL sketch data corrupted when read: unknown exception.");
+            }
+        }();
+
+        merge(cache);
+    }
+    double get_result() const {
+        if (hll_union_data.has_value()) {
+            try {
+                return hll_union_data->get_estimate();
+            } catch (const doris::Exception& e) {
+                throw Exception(e.code(),
+                                "Internal error happened when get HLL sketch estimate: {}",
+                                e.to_string());
+            } catch (const std::exception& e) {
+                throw Exception(ErrorCode::INTERNAL_ERROR,
+                                "Internal error happened when get HLL sketch estimate: {}",
+                                e.what());
+            } catch (...) {
+                throw Exception(
+                        ErrorCode::INTERNAL_ERROR,
+                        "Internal error happened when get HLL sketch estimate: unknown exception.");
+            }
+        }
+        return 0.0;
+    }
+};
+
+/// Calculates the number of different values approximately using hll sketch.
+template <PrimitiveType T, typename Data>
+class AggregateFunctionDataSketchesHllUnionAgg final
+        : public IAggregateFunctionDataHelper<Data,
+                                              AggregateFunctionDataSketchesHllUnionAgg<T, Data>>,
+          UnaryExpression,
+          NotNullableAggregateFunction {
+public:
+    AggregateFunctionDataSketchesHllUnionAgg(const DataTypes& argument_types_)
+            : IAggregateFunctionDataHelper<Data, AggregateFunctionDataSketchesHllUnionAgg<T, Data>>(
+                      argument_types_) {}
+    String get_name() const override { return Data::get_name(); }
+    DataTypePtr get_return_type() const override { return std::make_shared<DataTypeFloat64>(); }
+    void reset(AggregateDataPtr __restrict place) const override { this->data(place).reset(); }
+    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
+             Arena&) const override {
+        add_one(this->data(place), *columns[0], row_num);
+    }
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+               Arena&) const override {
+        const auto& rhs_data = this->data(rhs);
+        if (!rhs_data.hll_union_data.has_value()) {
+            return;
+        }
+        this->data(place).merge(rhs_data.hll_union_data->get_result(datasketches::HLL_8));
+    }
+    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
+        this->data(place).write(buf);
+    }
+    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+                     Arena&) const override {
+        this->data(place).read(buf);
+    }
+    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
+        assert_cast<ColumnFloat64&>(to).get_data().push_back(this->data(place).get_result());
+    }
+
+private:
+    static void ALWAYS_INLINE add_one(Data& data, const IColumn& column, ssize_t row_num) {
+        if constexpr (is_string_type(T) || is_varbinary(T)) {
+            const auto& src_column = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
+                                                 TypeCheckOnRelease::DISABLE>(column);
+            StringRef value = src_column.get_data_at(static_cast<size_t>(row_num));
+            if (value.empty()) {
+                throw Exception(ErrorCode::CORRUPTION,
+                                "HLL sketch data corrupted when add: empty input.");
+            }
+
+            using Sketch = typename Data::Sketch;
+            using Alloc = typename Data::Alloc;
+
+            auto sketch_data = [&]() -> Sketch {
+                try {
+                    return Sketch::deserialize(value.begin(), value.size, Alloc());
+                } catch (const doris::Exception& e) {
+                    throw Exception(e.code(), "Failed to deserialize HLL sketch when add: {}",
+                                    e.to_string());
+                } catch (const std::exception& e) {
+                    throw Exception(ErrorCode::CORRUPTION, "HLL sketch data corrupted when add: {}",
+                                    e.what());
+                } catch (...) {
+                    throw Exception(ErrorCode::CORRUPTION,
+                                    "HLL sketch data corrupted when add: unknown exception.");
+                }
+            }();
+
+            data.merge(sketch_data);
+        }
+    }
+};
+} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp b/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
index 9fc311cdf08380..6e18e95bee31f7 100644
--- a/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
@@ -38,6 +38,8 @@ void register_aggregate_function_avg(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_count(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_count_by_enum(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_HLL_union_agg(AggregateFunctionSimpleFactory& factory);
+void register_aggregate_function_datasketches_HLL_union_agg(
+        AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_uniq(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_uniq_distribute_key(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_bit(AggregateFunctionSimpleFactory& factory);
@@ -128,6 +130,7 @@ AggregateFunctionSimpleFactory& AggregateFunctionSimpleFactory::instance() {
         register_aggregate_function_replace_reader_load(instance);
         register_aggregate_function_window_lead_lag_first_last(instance);
         register_aggregate_function_HLL_union_agg(instance);
+        register_aggregate_function_datasketches_HLL_union_agg(instance);
         register_aggregate_functions_corr(instance);
         register_aggregate_functions_corr_welford(instance);
         register_aggregate_function_covar_pop(instance);
diff --git a/be/test/exprs/aggregate/agg_datasketches_hll_union_agg_test.cpp b/be/test/exprs/aggregate/agg_datasketches_hll_union_agg_test.cpp
new file mode 100644
index 00000000000000..7783ec89e2c832
--- /dev/null
+++ b/be/test/exprs/aggregate/agg_datasketches_hll_union_agg_test.cpp
@@ -0,0 +1,1097 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <DataSketches/hll.hpp>
+
+#include "agent/be_exec_version_manager.h"
+#include "common/config.h"
+#include "common/status.h"
+#include "core/column/column.h"
+#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_varbinary.h"
+#include "exec/common/hash_table/hash.h"
+#include "exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h"
+#include "exprs/aggregate/aggregate_function_simple_factory.h"
+
+namespace doris {
+
+void register_aggregate_function_datasketches_HLL_union_agg(
+        AggregateFunctionSimpleFactory& factory);
+
+class AggregateFunctionDataSketchesHllUnionAggTest : public ::testing::Test {
+protected:
+    void SetUp() override { arena = std::make_unique<Arena>(); }
+
+    void TearDown() override { arena.reset(); }
+
+    std::unique_ptr<Arena> arena;
+};
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testBasicUnion) {
+    // Create test: union multiple hll sketches and get correct cardinality estimate
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    // Create 2 different hll sketches, each with 100 unique values
+    datasketches::hll_sketch sketch1(8); // lg_k=8
+    for (int i = 0; i < 100; i++) {
+        sketch1.update(i);
+    }
+
+    datasketches::hll_sketch sketch2(8);
+    for (int i = 50; i < 150; i++) {
+        sketch2.update(i);
+    }
+
+    // Serialize both sketches into string column
+    auto column_string = ColumnString::create();
+    const auto ser1 = sketch1.serialize_compact();
+    column_string->insert_data((const char*)(ser1.data()), ser1.size());
+    const auto ser2 = sketch2.serialize_compact();
+    column_string->insert_data((const char*)(ser2.data()), ser2.size());
+
+    // Create aggregate data place
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    // Add both rows
+    const IColumn* columns[1] = {column_string.get()};
+    agg_func->add(place, columns, 0, *arena);
+    agg_func->add(place, columns, 1, *arena);
+
+    // Get result
+    ColumnFloat64 result_column;
+    agg_func->insert_result_into(place, result_column);
+
+    double estimate = result_column.get_data()[0];
+    EXPECT_GE(estimate, 130.0);
+    EXPECT_LE(estimate, 170.0);
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testMergeTwoAggStates) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    // Create two separate aggregate states
+    AggregateDataPtr place1 =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place1);
+
+    AggregateDataPtr place2 =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place2);
+
+    // Add different data to each state
+    datasketches::hll_sketch sketch1(8);
+    for (int i = 0; i < 100; i++) sketch1.update(i);
+    const auto ser1 = sketch1.serialize_compact();
+
+    datasketches::hll_sketch sketch2(8);
+    for (int i = 100; i < 200; i++) sketch2.update(i);
+    const auto ser2 = sketch2.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)(ser1.data()), ser1.size());
+    column_string->insert_data((const char*)(ser2.data()), ser2.size());
+
+    const IColumn* columns[1] = {column_string.get()};
+    agg_func->add(place1, columns, 0, *arena);
+    agg_func->add(place2, columns, 1, *arena);
+
+    // Merge second state into first
+    agg_func->merge(place1, place2, *arena);
+
+    // Get result
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place1, result);
+    double estimate = result.get_data()[0];
+
+    EXPECT_GE(estimate, 170.0);
+    EXPECT_LE(estimate, 230.0); // 200 unique values expected
+
+    agg_func->destroy(place1);
+    agg_func->destroy(place2);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testMergeEmptyStateDoesNotCrash) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    AggregateDataPtr place_with_data =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place_with_data);
+
+    AggregateDataPtr empty_rhs_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(empty_rhs_place);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 0; i < 100; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+    agg_func->add(place_with_data, columns, 0, *arena);
+
+    // Covers the "all NULL" style path: rhs exists but never saw add().
+    EXPECT_NO_THROW(agg_func->merge(place_with_data, empty_rhs_place, *arena));
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place_with_data, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], sketch.get_estimate());
+
+    // Empty string is invalid serialized sketch and should be rejected by add().
+    // Merge-empty-state coverage is handled by the "never saw add()" path above.
+
+    AggregateDataPtr empty_lhs_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(empty_lhs_place);
+    EXPECT_NO_THROW(agg_func->merge(empty_lhs_place, empty_rhs_place, *arena));
+
+    ColumnFloat64 empty_merge_result;
+    agg_func->insert_result_into(empty_lhs_place, empty_merge_result);
+    EXPECT_DOUBLE_EQ(empty_merge_result.get_data()[0], 0.0);
+
+    agg_func->destroy(place_with_data);
+    agg_func->destroy(empty_rhs_place);
+    agg_func->destroy(empty_lhs_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testEmptyState) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], 0.0);
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testSerializeDeserialize) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    // Add some data
+    datasketches::hll_sketch sketch(8);
+    for (int i = 0; i < 100; i++) sketch.update(i);
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)(ser.data()), ser.size());
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_string.get()};
+    agg_func->add(place, columns, 0, *arena);
+
+    // Serialize
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+    agg_func->serialize(place, w);
+    w.commit();
+
+    // Deserialize into new state
+    AggregateDataPtr new_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(new_place);
+
+    BufferReadable r(buffer->get_data_at(0));
+    agg_func->deserialize(new_place, r, *arena);
+
+    // Compare results
+    ColumnFloat64 result1, result2;
+    agg_func->insert_result_into(place, result1);
+    agg_func->insert_result_into(new_place, result2);
+
+    EXPECT_DOUBLE_EQ(result1.get_data()[0], result2.get_data()[0]);
+
+    agg_func->destroy(place);
+    agg_func->destroy(new_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testReset) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    datasketches::hll_sketch sketch(8);
+    for (int i = 0; i < 100; i++) sketch.update(i);
+    auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)(ser.data()), ser.size());
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_string.get()};
+    agg_func->add(place, columns, 0, *arena);
+
+    // Reset
+    agg_func->reset(place);
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], 0.0);
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testResetThenAddReinitializesState) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {input_type};
+
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    datasketches::hll_sketch sketch1(8, datasketches::HLL_8);
+    for (int i = 0; i < 7; ++i) {
+        sketch1.update(i);
+    }
+    const auto ser1 = sketch1.serialize_compact();
+
+    datasketches::hll_sketch sketch2(8, datasketches::HLL_8);
+    for (int i = 10; i < 17; ++i) {
+        sketch2.update(i);
+    }
+    const auto ser2 = sketch2.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser1.data(), ser1.size());
+    column_string->insert_data((const char*)ser2.data(), ser2.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    agg_func->add(place, columns, 0, *arena);
+    agg_func->reset(place);
+    agg_func->add(place, columns, 1, *arena);
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], sketch2.get_estimate());
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest,
+       testVarcharResetThenAddAndMergeEmptyRhsDoesNotCrash) {
+    DataTypePtr input_type = std::make_shared<DataTypeString>();
+    DataTypes argument_types = {std::make_shared<DataTypeString>(-1, TYPE_VARCHAR)};
+
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_VARCHAR,
+                                                     AggregateFunctionHllSketchData<TYPE_VARCHAR>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    datasketches::hll_sketch sketch1(8, datasketches::HLL_8);
+    for (int i = 0; i < 7; ++i) {
+        sketch1.update(i);
+    }
+    const auto ser1 = sketch1.serialize_compact();
+
+    datasketches::hll_sketch sketch2(8, datasketches::HLL_8);
+    for (int i = 10; i < 17; ++i) {
+        sketch2.update(i);
+    }
+    const auto ser2 = sketch2.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser1.data(), ser1.size());
+    column_string->insert_data((const char*)ser2.data(), ser2.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    agg_func->add(place, columns, 0, *arena);
+    agg_func->reset(place);
+    agg_func->add(place, columns, 1, *arena);
+
+    ColumnFloat64 after_reset;
+    agg_func->insert_result_into(place, after_reset);
+    EXPECT_DOUBLE_EQ(after_reset.get_data()[0], sketch2.get_estimate());
+
+    AggregateDataPtr empty_rhs_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(empty_rhs_place);
+
+    EXPECT_NO_THROW(agg_func->merge(place, empty_rhs_place, *arena));
+
+    ColumnFloat64 after_merge;
+    agg_func->insert_result_into(place, after_merge);
+    EXPECT_DOUBLE_EQ(after_merge.get_data()[0], sketch2.get_estimate());
+
+    AggregateDataPtr empty_lhs_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(empty_lhs_place);
+
+    EXPECT_NO_THROW(agg_func->merge(empty_lhs_place, empty_rhs_place, *arena));
+
+    ColumnFloat64 empty_merge_result;
+    agg_func->insert_result_into(empty_lhs_place, empty_merge_result);
+    EXPECT_DOUBLE_EQ(empty_merge_result.get_data()[0], 0.0);
+
+    agg_func->destroy(place);
+    agg_func->destroy(empty_rhs_place);
+    agg_func->destroy(empty_lhs_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testFactoryCreateAndAliases) {
+    AggregateFunctionSimpleFactory factory;
+    register_aggregate_function_datasketches_HLL_union_agg(factory);
+    int be_version = BeExecVersionManager::get_newest_version();
+
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+
+    auto fn_main =
+            factory.get("datasketches_hll_union_agg", argument_types, nullptr, false, be_version);
+    auto fn_alias_sr_estimate =
+            factory.get("ds_hll_estimate", argument_types, nullptr, false, be_version);
+    auto fn_alias_datasketches_estimate =
+            factory.get("datasketches_hll_estimate", argument_types, nullptr, false, be_version);
+
+    ASSERT_NE(fn_main, nullptr);
+    ASSERT_NE(fn_alias_sr_estimate, nullptr);
+    ASSERT_NE(fn_alias_datasketches_estimate, nullptr);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 0; i < 7; ++i) sketch.update(i);
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    auto run_and_get_result = [&](const AggregateFunctionPtr& fn) {
+        AggregateDataPtr place = arena->aligned_alloc(fn->size_of_data(), fn->align_of_data());
+        fn->create(place);
+        fn->add(place, columns, 0, *arena);
+        ColumnFloat64 result;
+        fn->insert_result_into(place, result);
+        fn->destroy(place);
+        return result.get_data()[0];
+    };
+
+    double expected = sketch.get_estimate();
+    EXPECT_DOUBLE_EQ(run_and_get_result(fn_main), expected);
+    EXPECT_DOUBLE_EQ(run_and_get_result(fn_alias_sr_estimate), expected);
+    EXPECT_DOUBLE_EQ(run_and_get_result(fn_alias_datasketches_estimate), expected);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest,
+       testFactoryCreateForVarcharAndNullableAndUnsupportedType) {
+    AggregateFunctionSimpleFactory factory;
+    register_aggregate_function_datasketches_HLL_union_agg(factory);
+    int be_version = BeExecVersionManager::get_newest_version();
+
+    DataTypes varchar_types = {std::make_shared<DataTypeString>(-1, TYPE_VARCHAR)};
+    auto fn_varchar =
+            factory.get("datasketches_hll_union_agg", varchar_types, nullptr, false, be_version);
+    auto fn_varchar_alias =
+            factory.get("ds_hll_estimate", varchar_types, nullptr, false, be_version);
+    ASSERT_NE(fn_varchar, nullptr);
+    ASSERT_NE(fn_varchar_alias, nullptr);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 0; i < 7; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    auto run_and_get_result = [&](const AggregateFunctionPtr& fn) {
+        AggregateDataPtr place = arena->aligned_alloc(fn->size_of_data(), fn->align_of_data());
+        fn->create(place);
+        fn->add(place, columns, 0, *arena);
+        ColumnFloat64 result;
+        fn->insert_result_into(place, result);
+        fn->destroy(place);
+        return result.get_data()[0];
+    };
+
+    double expected = sketch.get_estimate();
+    EXPECT_DOUBLE_EQ(run_and_get_result(fn_varchar), expected);
+    EXPECT_DOUBLE_EQ(run_and_get_result(fn_varchar_alias), expected);
+
+    DataTypes nullable_varchar_types = {
+            make_nullable(std::make_shared<DataTypeString>(-1, TYPE_VARCHAR))};
+    auto fn_nullable_varchar = factory.get("datasketches_hll_union_agg", nullable_varchar_types,
+                                           nullptr, false, be_version);
+    ASSERT_NE(fn_nullable_varchar, nullptr);
+
+    DataTypes unsupported_types = {std::make_shared<DataTypeInt32>()};
+    auto fn_unsupported = factory.get("datasketches_hll_union_agg", unsupported_types, nullptr,
+                                      false, be_version);
+    EXPECT_EQ(fn_unsupported, nullptr);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testLowLgKSketchDoesNotReportCorruption) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    datasketches::hll_sketch sketch(4, datasketches::HLL_8);
+    for (int i = 0; i < 100; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_string.get()};
+    EXPECT_NO_THROW(agg_func->add(place, columns, 0, *arena));
+
+    ColumnFloat64 add_result;
+    agg_func->insert_result_into(place, add_result);
+
+    EXPECT_GE(add_result.get_data()[0], 50);
+    EXPECT_LE(add_result.get_data()[0], 150);
+
+    auto buf = ColumnString::create();
+    BufferWritable w(*buf);
+    StringRef d((const char*)ser.data(), ser.size());
+    w.write_binary(d);
+    w.commit();
+
+    AggregateDataPtr place2 =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place2);
+
+    BufferReadable r(buf->get_data_at(0));
+    EXPECT_NO_THROW(agg_func->deserialize(place2, r, *arena));
+
+    ColumnFloat64 deserialize_result;
+    agg_func->insert_result_into(place2, deserialize_result);
+    EXPECT_EQ(deserialize_result.get_data()[0], add_result.get_data()[0]);
+
+    agg_func->destroy(place);
+    agg_func->destroy(place2);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testAddEmptyStringThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data("", 0);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_string.get()};
+
+    try {
+        agg_func->add(place, columns, 0, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testResetOnEmptyState) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    EXPECT_NO_THROW(agg_func->reset(place)); // cover reset() branch when union_data is nullptr
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], 0.0);
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarbinaryInput) {
+    AggregateFunctionSimpleFactory factory;
+    register_aggregate_function_datasketches_HLL_union_agg(factory);
+    int be_version = BeExecVersionManager::get_newest_version();
+
+    DataTypes argument_types = {std::make_shared<DataTypeVarbinary>()};
+    auto fn = factory.get("datasketches_hll_union_agg", argument_types, nullptr, false, be_version);
+    ASSERT_NE(fn, nullptr);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 20; i < 30; ++i) sketch.update(i);
+    const auto ser = sketch.serialize_compact();
+
+    auto column_varbinary = ColumnVarbinary::create();
+    column_varbinary->insert_data((const char*)ser.data(), ser.size());
+
+    const IColumn* columns[1] = {column_varbinary.get()};
+
+    AggregateDataPtr place = arena->aligned_alloc(fn->size_of_data(), fn->align_of_data());
+    fn->create(place);
+    fn->add(place, columns, 0, *arena);
+
+    ColumnFloat64 result;
+    fn->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], sketch.get_estimate());
+
+    fn->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarbinaryAddEmptyStringThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeVarbinary>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARBINARY, AggregateFunctionHllSketchData<TYPE_VARBINARY>>>(argument_types);
+
+    auto column_varbinary = ColumnVarbinary::create();
+    column_varbinary->insert_data("", 0);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_varbinary.get()};
+
+    try {
+        agg_func->add(place, columns, 0, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarbinaryCorruptedInputThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeVarbinary>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARBINARY, AggregateFunctionHllSketchData<TYPE_VARBINARY>>>(argument_types);
+
+    auto column_varbinary = ColumnVarbinary::create();
+    column_varbinary->insert_data("x", 1);
+    const IColumn* columns[1] = {column_varbinary.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    try {
+        agg_func->add(place, columns, 0, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    auto corrupt_buf = ColumnString::create();
+    BufferWritable corrupt_w(*corrupt_buf);
+    StringRef corrupted("x", 1);
+    corrupt_w.write_binary(corrupted);
+    corrupt_w.commit();
+
+    BufferReadable r(corrupt_buf->get_data_at(0));
+    try {
+        agg_func->deserialize(place, r, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarbinarySerializeDeserialize) {
+    DataTypes argument_types = {std::make_shared<DataTypeVarbinary>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARBINARY, AggregateFunctionHllSketchData<TYPE_VARBINARY>>>(argument_types);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 0; i < 100; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_varbinary = ColumnVarbinary::create();
+    column_varbinary->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_varbinary.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+    agg_func->add(place, columns, 0, *arena);
+
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+    agg_func->serialize(place, w);
+    w.commit();
+
+    AggregateDataPtr new_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(new_place);
+
+    BufferReadable r(buffer->get_data_at(0));
+    agg_func->deserialize(new_place, r, *arena);
+
+    ColumnFloat64 result1, result2;
+    agg_func->insert_result_into(place, result1);
+    agg_func->insert_result_into(new_place, result2);
+
+    EXPECT_DOUBLE_EQ(result1.get_data()[0], result2.get_data()[0]);
+
+    agg_func->destroy(place);
+    agg_func->destroy(new_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testSerializeDeserializeEmptyState) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+    EXPECT_NO_THROW(agg_func->serialize(place, w)); // covers write() empty-state branch
+    w.commit();
+
+    AggregateDataPtr new_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(new_place);
+
+    BufferReadable r(buffer->get_data_at(0));
+    agg_func->deserialize(new_place, r, *arena);
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(new_place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], 0.0);
+
+    agg_func->destroy(place);
+    agg_func->destroy(new_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testCorruptedInputThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_STRING, AggregateFunctionHllSketchData<TYPE_STRING>>>(argument_types);
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data("x", 1); // definitely not a valid sketch
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    try {
+        agg_func->add(place, columns, 0, *arena); // covers add() CORRUPTION catch branch
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    auto corrupt_buf = ColumnString::create();
+    BufferWritable corrupt_w(*corrupt_buf);
+    StringRef corrupted("x", 1);
+    corrupt_w.write_binary(corrupted);
+    corrupt_w.commit();
+
+    BufferReadable r(corrupt_buf->get_data_at(0));
+    try {
+        agg_func->deserialize(place, r, *arena); // covers read() CORRUPTION catch branch
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testAllocatorAwareSketchInput) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    using Alloc = doris::CustomStdAllocator<uint8_t>;
+    using Sketch = datasketches::hll_sketch_alloc<Alloc>;
+
+    Sketch sketch(8, datasketches::HLL_8, false, Alloc());
+    for (int i = 0; i < 7; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    agg_func->add(place, columns, 0, *arena);
+
+    ColumnFloat64 result;
+    agg_func->insert_result_into(place, result);
+    EXPECT_DOUBLE_EQ(result.get_data()[0], sketch.get_estimate());
+
+    agg_func->destroy(place);
+}
+
+namespace {
+
+class ScopedMemAllocFaultInjection {
+public:
+    explicit ScopedMemAllocFaultInjection(double probability)
+            : _old_probability(doris::config::mem_alloc_fault_probability) {
+        doris::config::mem_alloc_fault_probability = probability;
+        doris::enable_thread_catch_bad_alloc++;
+    }
+
+    ~ScopedMemAllocFaultInjection() {
+        doris::enable_thread_catch_bad_alloc--;
+        doris::config::mem_alloc_fault_probability = _old_probability;
+    }
+
+    ScopedMemAllocFaultInjection(const ScopedMemAllocFaultInjection&) = delete;
+    ScopedMemAllocFaultInjection& operator=(const ScopedMemAllocFaultInjection&) = delete;
+
+private:
+    double _old_probability;
+};
+
+} // namespace
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testMetaInfoCoversInlineMethods) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    EXPECT_EQ(AggregateFunctionHllSketchData<TYPE_STRING>::get_name(),
+              "datasketches_hll_union_agg");
+    EXPECT_EQ(agg_func->get_name(), "datasketches_hll_union_agg");
+
+    auto return_type = agg_func->get_return_type();
+    EXPECT_TRUE(return_type->equals(*std::make_shared<DataTypeFloat64>()));
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testAddMemAllocFailedThrowsMemAllocFailed) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    datasketches::hll_sketch sketch(12, datasketches::HLL_8);
+    for (int i = 0; i < 1000; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    {
+        ScopedMemAllocFaultInjection inject(1.0);
+        try {
+            agg_func->add(place, columns, 0, *arena);
+            FAIL() << "Expected doris::Exception";
+        } catch (const doris::Exception& e) {
+            EXPECT_EQ(e.code(), doris::ErrorCode::MEM_ALLOC_FAILED);
+        }
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest,
+       testDeserializeMemAllocFailedThrowsMemAllocFailed) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    datasketches::hll_sketch sketch(12, datasketches::HLL_8);
+    for (int i = 0; i < 1000; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+    StringRef d((const char*)ser.data(), ser.size());
+    w.write_binary(d);
+    w.commit();
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    BufferReadable r(buffer->get_data_at(0));
+    {
+        ScopedMemAllocFaultInjection inject(1.0);
+        try {
+            agg_func->deserialize(place, r, *arena);
+            FAIL() << "Expected doris::Exception";
+        } catch (const doris::Exception& e) {
+            EXPECT_EQ(e.code(), doris::ErrorCode::MEM_ALLOC_FAILED);
+        }
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarcharAddEmptyStringThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>(-1, TYPE_VARCHAR)};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARCHAR, AggregateFunctionHllSketchData<TYPE_VARCHAR>>>(argument_types);
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data("", 0);
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    const IColumn* columns[1] = {column_string.get()};
+
+    try {
+        agg_func->add(place, columns, 0, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarcharCorruptedInputThrows) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>(-1, TYPE_VARCHAR)};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARCHAR, AggregateFunctionHllSketchData<TYPE_VARCHAR>>>(argument_types);
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data("x", 1);
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+
+    try {
+        agg_func->add(place, columns, 0, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    auto corrupt_buf = ColumnString::create();
+    BufferWritable corrupt_w(*corrupt_buf);
+    StringRef corrupted("x", 1);
+    corrupt_w.write_binary(corrupted);
+    corrupt_w.commit();
+
+    BufferReadable r(corrupt_buf->get_data_at(0));
+    try {
+        agg_func->deserialize(place, r, *arena);
+        FAIL() << "Expected doris::Exception";
+    } catch (const doris::Exception& e) {
+        EXPECT_EQ(e.code(), doris::ErrorCode::CORRUPTION);
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testVarcharSerializeDeserialize) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>(-1, TYPE_VARCHAR)};
+    auto agg_func = std::make_shared<AggregateFunctionDataSketchesHllUnionAgg<
+            TYPE_VARCHAR, AggregateFunctionHllSketchData<TYPE_VARCHAR>>>(argument_types);
+
+    datasketches::hll_sketch sketch(8, datasketches::HLL_8);
+    for (int i = 0; i < 100; i++) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+    agg_func->add(place, columns, 0, *arena);
+
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+    agg_func->serialize(place, w);
+    w.commit();
+
+    AggregateDataPtr new_place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(new_place);
+
+    BufferReadable r(buffer->get_data_at(0));
+    agg_func->deserialize(new_place, r, *arena);
+
+    ColumnFloat64 result1, result2;
+    agg_func->insert_result_into(place, result1);
+    agg_func->insert_result_into(new_place, result2);
+
+    EXPECT_DOUBLE_EQ(result1.get_data()[0], result2.get_data()[0]);
+
+    agg_func->destroy(place);
+    agg_func->destroy(new_place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest,
+       testSerializeMemAllocFailedThrowsMemAllocFailed) {
+    DataTypes argument_types = {std::make_shared<DataTypeString>()};
+    using AggFunc =
+            AggregateFunctionDataSketchesHllUnionAgg<TYPE_STRING,
+                                                     AggregateFunctionHllSketchData<TYPE_STRING>>;
+    auto agg_func = std::make_shared<AggFunc>(argument_types);
+
+    datasketches::hll_sketch sketch(12, datasketches::HLL_8);
+    for (int i = 0; i < 2000; ++i) {
+        sketch.update(i);
+    }
+    const auto ser = sketch.serialize_compact();
+
+    auto column_string = ColumnString::create();
+    column_string->insert_data((const char*)ser.data(), ser.size());
+    const IColumn* columns[1] = {column_string.get()};
+
+    AggregateDataPtr place =
+            arena->aligned_alloc(agg_func->size_of_data(), agg_func->align_of_data());
+    agg_func->create(place);
+    agg_func->add(place, columns, 0, *arena);
+
+    auto buffer = ColumnString::create();
+    BufferWritable w(*buffer);
+
+    {
+        ScopedMemAllocFaultInjection inject(1.0);
+        try {
+            agg_func->serialize(place, w);
+            FAIL() << "Expected doris::Exception";
+        } catch (const doris::Exception& e) {
+            EXPECT_EQ(e.code(), doris::ErrorCode::MEM_ALLOC_FAILED);
+            EXPECT_NE(e.to_string().find("serialize HLL sketch"), std::string::npos);
+        }
+    }
+
+    agg_func->destroy(place);
+}
+
+TEST_F(AggregateFunctionDataSketchesHllUnionAggTest, testDataMergeDownsampleMemAllocFailed) {
+    using Data = AggregateFunctionHllSketchData<TYPE_STRING>;
+    using Alloc = Data::Alloc;
+    using Sketch = Data::Sketch;
+
+    Data data;
+
+    Sketch large_k_sketch(12, datasketches::HLL_8, true, Alloc());
+    for (int i = 0; i < 5000; ++i) {
+        large_k_sketch.update(i);
+    }
+    EXPECT_NO_THROW(data.merge(large_k_sketch));
+
+    Sketch small_k_sketch(8, datasketches::HLL_8, true, Alloc());
+    for (int i = 0; i < 5000; ++i) {
+        small_k_sketch.update(i);
+    }
+
+    {
+        ScopedMemAllocFaultInjection inject(1.0);
+        try {
+            data.merge(small_k_sketch);
+            FAIL() << "Expected doris::Exception";
+        } catch (const doris::Exception& e) {
+            EXPECT_EQ(e.code(), doris::ErrorCode::MEM_ALLOC_FAILED);
+            EXPECT_NE(e.to_string().find("update HLL sketch"), std::string::npos);
+        }
+    }
+}
+
+} // namespace doris
diff --git a/build.sh b/build.sh
index 66c9836cf0a620..c0cba232b2d16b 100755
--- a/build.sh
+++ b/build.sh
@@ -34,6 +34,7 @@ if [[ -z "${DORIS_THIRDPARTY}" ]]; then
     export DORIS_THIRDPARTY="${DORIS_HOME}/thirdparty"
 fi
 export TP_INCLUDE_DIR="${DORIS_THIRDPARTY}/installed/include"
+export TP_INSTALLED_DIR="${DORIS_THIRDPARTY}/installed"
 export TP_LIB_DIR="${DORIS_THIRDPARTY}/installed/lib"
 HADOOP_DEPS_NAME="hadoop-deps"
 . "${DORIS_HOME}/env.sh"
@@ -628,6 +629,14 @@ FE_MODULES="$(
 
 # Clean and build Backend
 if [[ "${BUILD_BE}" -eq 1 ]]; then
+
+    echo "install datasketches-cpp to thirdparty path before build be"
+    update_submodule "contrib/datasketches-cpp" "datasketches-cpp" "https://github.com/apache/datasketches-cpp/archive/refs/heads/master.tar.gz"
+    cd "${DORIS_HOME}/contrib/datasketches-cpp"
+    "${CMAKE_CMD}" -S . -B build/Release -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=$TP_INSTALLED_DIR -DBUILD_TESTS=OFF
+    "${CMAKE_CMD}" --build build/Release -t install
+    cd "${DORIS_HOME}"
+
     update_submodule "contrib/apache-orc" "apache-orc" "https://github.com/apache/doris-thirdparty/archive/refs/heads/orc.tar.gz"
     update_submodule "contrib/clucene" "clucene" "https://github.com/apache/doris-thirdparty/archive/refs/heads/clucene.tar.gz"
     update_submodule "contrib/openblas" "openblas" "https://github.com/apache/doris-thirdparty/archive/refs/heads/openblas.tar.gz"
diff --git a/contrib/datasketches-cpp b/contrib/datasketches-cpp
new file mode 160000
index 00000000000000..de8553ba372e61
--- /dev/null
+++ b/contrib/datasketches-cpp
@@ -0,0 +1 @@
+Subproject commit de8553ba372e618382c2e7b44b0ffc9422b9458c
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
index 66140acc790d91..8d08bba9ce72e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
@@ -38,6 +38,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
 import org.apache.doris.nereids.trees.expressions.functions.agg.CovarSamp;
+import org.apache.doris.nereids.trees.expressions.functions.agg.DataSketchesHllUnionAgg;
 import org.apache.doris.nereids.trees.expressions.functions.agg.ExponentialMovingAverage;
 import org.apache.doris.nereids.trees.expressions.functions.agg.GroupArrayIntersect;
 import org.apache.doris.nereids.trees.expressions.functions.agg.GroupArrayUnion;
@@ -148,6 +149,8 @@ private BuiltinAggregateFunctions() {
                 agg(Histogram.class, "hist", "histogram"),
                 agg(HllUnion.class, "hll_raw_agg", "hll_union"),
                 agg(HllUnionAgg.class, "hll_union_agg"),
+                agg(DataSketchesHllUnionAgg.class, "datasketches_hll_union_agg",
+                    "ds_hll_estimate", "datasketches_hll_estimate"),
                 agg(IntersectCount.class, "intersect_count"),
                 agg(Kurt.class, "kurt", "kurt_pop", "kurtosis"),
                 agg(LinearHistogram.class, "linear_histogram"),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/DataSketchesHllUnionAgg.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/DataSketchesHllUnionAgg.java
new file mode 100644
index 00000000000000..9fe46d2a77dd45
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/DataSketchesHllUnionAgg.java
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.agg;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.FunctionTrait;
+import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DoubleType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VarBinaryType;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/** datasketches_hll_union_agg agg function. */
+public class DataSketchesHllUnionAgg extends NotNullableAggregateFunction
+        implements UnaryExpression, ExplicitlyCastableSignature, FunctionTrait, RollUpTrait {
+    public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+            FunctionSignature.ret(DoubleType.INSTANCE).args(StringType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(VarBinaryType.INSTANCE)
+    );
+
+    /**
+     * constructor with 1 argument.
+     */
+    public DataSketchesHllUnionAgg(Expression arg) {
+        super("datasketches_hll_union_agg", arg);
+    }
+
+    /**
+     * constructor with 1 argument.
+     */
+    public DataSketchesHllUnionAgg(boolean distinct, Expression arg) {
+        this(arg);
+    }
+
+    /** constructor for withChildren and reuse signature */
+    protected DataSketchesHllUnionAgg(AggregateFunctionParams functionParams) {
+        super(functionParams);
+    }
+
+    @Override
+    public void checkLegalityBeforeTypeCoercion() {
+        DataType inputType = getArgumentType(0);
+        if (!(inputType.isStringType() || inputType.isVarcharType() || inputType.isVarBinaryType()
+                || inputType.isNullType())) {
+            throw new AnalysisException(getName()
+                + " function's argument should be of STRING/VARCHAR/VARBINARY type, but was " + inputType);
+        }
+    }
+
+    @Override
+    protected List<DataType> intermediateTypes() {
+        return ImmutableList.of(StringType.INSTANCE);
+    }
+
+    @Override
+    public List<FunctionSignature> getSignatures() {
+        return SIGNATURES;
+    }
+
+    @Override
+    public DataSketchesHllUnionAgg withDistinctAndChildren(boolean distinct, List<Expression> children) {
+        Preconditions.checkArgument(children.size() == 1);
+        return new DataSketchesHllUnionAgg(getFunctionParams(distinct, children));
+    }
+
+    @Override
+    public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+        return visitor.visitDataSketchesHllUnionAgg(this, context);
+    }
+
+    @Override
+    public Function constructRollUp(Expression param, Expression... varParams) {
+        return new DataSketchesHllUnionAgg(getFunctionParams(ImmutableList.of(param)));
+    }
+
+    @Override
+    public boolean canRollUp() {
+        return false;
+    }
+
+    @Override
+    public Expression resultForEmptyInput() {
+        return new DoubleLiteral(0);
+    }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
index 251012c16fec5e..5292a967c27343 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
@@ -39,6 +39,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
 import org.apache.doris.nereids.trees.expressions.functions.agg.CovarSamp;
+import org.apache.doris.nereids.trees.expressions.functions.agg.DataSketchesHllUnionAgg;
 import org.apache.doris.nereids.trees.expressions.functions.agg.ExponentialMovingAverage;
 import org.apache.doris.nereids.trees.expressions.functions.agg.GroupArrayIntersect;
 import org.apache.doris.nereids.trees.expressions.functions.agg.GroupArrayUnion;
@@ -238,6 +239,10 @@ default R visitHistogram(Histogram histogram, C context) {
         return visitAggregateFunction(histogram, context);
     }
 
+    default R visitDataSketchesHllUnionAgg(DataSketchesHllUnionAgg datasketchesHllUnionAgg, C context) {
+        return visitAggregateFunction(datasketchesHllUnionAgg, context);
+    }
+
     default R visitHllUnion(HllUnion hllUnion, C context) {
         return visitAggregateFunction(hllUnion, context);
     }
diff --git a/regression-test/data/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.out b/regression-test/data/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.out
new file mode 100644
index 00000000000000..ae1424305c7178
--- /dev/null
+++ b/regression-test/data/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.out
@@ -0,0 +1,28 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !basic_union --
+17
+
+-- !aliases --
+17	17	17
+
+-- !group_by --
+1	7
+2	10
+
+-- !distinct --
+17	17
+
+-- !basic_union_varchar --
+17
+
+-- !aliases_varchar --
+17	17	17
+
+-- !basic_union_varbinary --
+17
+
+-- !aliases_varbinary --
+17	17	17
+
+-- !empty_input --
+0
diff --git a/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.groovy b/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.groovy
new file mode 100644
index 00000000000000..8d86b8c82d6def
--- /dev/null
+++ b/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_datasketches_hll_union_agg.groovy
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_datasketches_hll_union_agg") {
+    def tableName = "test_datasketches_hll_union_agg_tbl"
+    def varcharTableName = "test_datasketches_hll_union_agg_varchar_tbl"
+    def emptyTableName = "test_datasketches_hll_union_agg_empty_tbl"
+    def badTableName = "test_datasketches_hll_union_agg_bad_tbl"
+
+    // sk = new HllSketch(8, HLL_8); for (int i = 0; i < 7; i++) sk.update(i);
+    def sk1Base64 = "AgEHCAMIBwjL18IEK/L7BoYv+Q11gWYHgbxdBntl5gj8LUIK"
+
+    // sk = new HllSketch(8, HLL_8); for (int i = 20; i < 30; i++) sk.update(i);
+    def sk2Base64 = "AwEHCAUIAAkKAAAAIjvrBcS1nwfGGWoEyHokBO8t9wc1qTEENkcJB7hWqQxZf9QNnuSbGA=="
+
+    sql "DROP TABLE IF EXISTS ${tableName}"
+    sql """
+        CREATE TABLE ${tableName} (
+            id INT,
+            sk STRING
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_num" = "1"
+        )
+    """
+
+    sql """
+        INSERT INTO ${tableName} VALUES
+            (1, from_base64('${sk1Base64}')),
+            (2, from_base64('${sk2Base64}')),
+            (3, NULL)
+    """
+
+    // 1) Basic union: {0..6} U {20..29} => 17 distinct values
+    qt_basic_union """SELECT CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT) FROM ${tableName}"""
+
+    // 2) Aliases should behave identically
+    qt_aliases """SELECT
+            CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT),
+            CAST(ROUND(ds_hll_estimate(sk)) AS BIGINT),
+            CAST(ROUND(datasketches_hll_estimate(sk)) AS BIGINT)
+        FROM ${tableName}
+    """
+
+    // 3) Group-by
+    qt_group_by """SELECT id, CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT)
+        FROM ${tableName}
+        WHERE id IN (1, 2)
+        GROUP BY id
+        ORDER BY id
+    """
+
+    // 4) DISTINCT should not change result in this data set
+    sql "INSERT INTO ${tableName} VALUES (5, from_base64('${sk1Base64}'))"
+    qt_distinct """SELECT
+            CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT),
+            CAST(ROUND(datasketches_hll_union_agg(DISTINCT sk)) AS BIGINT)
+        FROM ${tableName}
+    """
+
+    // 4.1) Input type coverage: VARCHAR
+    sql "DROP TABLE IF EXISTS ${varcharTableName}"
+    sql """
+        CREATE TABLE ${varcharTableName} (
+            id INT,
+            sk VARCHAR(256)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_num" = "1"
+        )
+    """
+
+    sql """
+        INSERT INTO ${varcharTableName} VALUES
+            (1, from_base64('${sk1Base64}')),
+            (2, from_base64('${sk2Base64}')),
+            (3, NULL)
+    """
+
+    qt_basic_union_varchar """SELECT CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT) FROM ${varcharTableName}"""
+
+    qt_aliases_varchar """SELECT
+            CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT),
+            CAST(ROUND(ds_hll_estimate(sk)) AS BIGINT),
+            CAST(ROUND(datasketches_hll_estimate(sk)) AS BIGINT)
+        FROM ${varcharTableName}
+    """
+
+    // 4.2) Input type coverage: VARBINARY (no table column; VARBINARY is not supported for table storage)
+    qt_basic_union_varbinary """SELECT CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT) FROM (
+            SELECT from_base64_binary('${sk1Base64}') AS sk
+            UNION ALL SELECT from_base64_binary('${sk2Base64}')
+            UNION ALL SELECT NULL
+        ) t
+    """
+
+    qt_aliases_varbinary """SELECT
+            CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT),
+            CAST(ROUND(ds_hll_estimate(sk)) AS BIGINT),
+            CAST(ROUND(datasketches_hll_estimate(sk)) AS BIGINT)
+        FROM (
+            SELECT from_base64_binary('${sk1Base64}') AS sk
+            UNION ALL SELECT from_base64_binary('${sk2Base64}')
+            UNION ALL SELECT NULL
+        ) t
+    """
+
+    // 5) Empty input should return 0
+    sql "DROP TABLE IF EXISTS ${emptyTableName}"
+    sql """
+        CREATE TABLE ${emptyTableName} (
+            id INT,
+            sk STRING
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_num" = "1"
+        )
+    """
+    qt_empty_input """SELECT CAST(ROUND(datasketches_hll_union_agg(sk)) AS BIGINT) FROM ${emptyTableName}"""
+
+    // 6) Illegal input should throw (base64 is valid but bytes are not a datasketches HLL sketch)
+    test {
+        sql """SELECT datasketches_hll_union_agg(from_base64('AA=='))"""
+        exception "CORRUPTION"
+    }
+    test {
+        sql """SELECT ds_hll_estimate(from_base64('AA=='))"""
+        exception "CORRUPTION"
+    }
+    test {
+        sql """SELECT datasketches_hll_estimate(from_base64('AA=='))"""
+        exception "CORRUPTION"
+    }
+
+    // Empty string is a valid STRING value, but it is an invalid serialized DataSketches HLL sketch.
+    // It should not fail at INSERT time; it should fail when the aggregate function reads it.
+    sql "DROP TABLE IF EXISTS ${badTableName}"
+    sql """
+        CREATE TABLE ${badTableName} (
+            id INT,
+            sk STRING
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_num" = "1"
+        )
+    """
+    sql """INSERT INTO ${badTableName} VALUES (1, '')"""
+    test {
+        sql """SELECT datasketches_hll_union_agg(sk) FROM ${badTableName}"""
+        exception "CORRUPTION"
+    }
+}
diff --git a/run-be-ut.sh b/run-be-ut.sh
index c9579ead354bd8..1779edf05d728f 100755
--- a/run-be-ut.sh
+++ b/run-be-ut.sh
@@ -41,7 +41,12 @@ ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
 
 export ROOT
 export DORIS_HOME="${ROOT}"
-
+if [[ -z "${DORIS_THIRDPARTY}" ]]; then
+    export DORIS_THIRDPARTY="${DORIS_HOME}/thirdparty"
+fi
+export TP_INCLUDE_DIR="${DORIS_THIRDPARTY}/installed/include"
+export TP_INSTALLED_DIR="${DORIS_THIRDPARTY}/installed"
+export TP_LIB_DIR="${DORIS_THIRDPARTY}/installed/lib"
 . "${DORIS_HOME}/env.sh"
 
 # Check args
@@ -174,6 +179,13 @@ update_submodule() {
     fi
 }
 
+echo "install datasketches-cpp to thirdparty path before build backend ut"
+update_submodule "contrib/datasketches-cpp" "datasketches-cpp" "https://github.com/apache/datasketches-cpp/archive/refs/heads/master.tar.gz"
+cd "${DORIS_HOME}/contrib/datasketches-cpp"
+"${CMAKE_CMD}" -S . -B build/Release -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=$TP_INSTALLED_DIR -DBUILD_TESTS=OFF
+"${CMAKE_CMD}" --build build/Release -t install
+cd "${DORIS_HOME}"
+
 update_submodule "contrib/apache-orc" "apache-orc" "https://github.com/apache/doris-thirdparty/archive/refs/heads/orc.tar.gz"
 update_submodule "contrib/clucene" "clucene" "https://github.com/apache/doris-thirdparty/archive/refs/heads/clucene.tar.gz"