diff --git a/.gitmodules b/.gitmodules
index 6acf8b86ad42..343e61166d95 100644
--- a/.gitmodules
+++ b/.gitmodules
@@ -154,7 +154,7 @@
 	url = https://github.com/ClickHouse/NuRaft
 [submodule "contrib/datasketches-cpp"]
 	path = contrib/datasketches-cpp
-	url = https://github.com/apache/datasketches-cpp
+	url = https://github.com/ClickHouse/datasketches-cpp
 [submodule "contrib/yaml-cpp"]
 	path = contrib/yaml-cpp
 	url = https://github.com/ClickHouse/yaml-cpp
diff --git a/contrib/datasketches-cpp b/contrib/datasketches-cpp
index 76edd74f5db2..23bd9b070ddf 160000
--- a/contrib/datasketches-cpp
+++ b/contrib/datasketches-cpp
@@ -1 +1 @@
-Subproject commit 76edd74f5db286b672c170a8ded4ce39b3a8800f
+Subproject commit 23bd9b070ddf7f0dcf32c87d6752ce535374697b
diff --git a/contrib/datasketches-cpp-cmake/CMakeLists.txt b/contrib/datasketches-cpp-cmake/CMakeLists.txt
index 497d6956d0ef..ff4199c17430 100644
--- a/contrib/datasketches-cpp-cmake/CMakeLists.txt
+++ b/contrib/datasketches-cpp-cmake/CMakeLists.txt
@@ -10,6 +10,7 @@ add_library(_datasketches INTERFACE)
 target_include_directories(_datasketches SYSTEM BEFORE INTERFACE
     "${ClickHouse_SOURCE_DIR}/contrib/datasketches-cpp/common/include"
     "${ClickHouse_SOURCE_DIR}/contrib/datasketches-cpp/count/include"
+    "${ClickHouse_SOURCE_DIR}/contrib/datasketches-cpp/hll/include"
     "${ClickHouse_SOURCE_DIR}/contrib/datasketches-cpp/theta/include")
 
 add_library(ch_contrib::datasketches ALIAS _datasketches)
diff --git a/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.cpp b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.cpp
new file mode 100644
index 000000000000..21fb8363e155
--- /dev/null
+++ b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.cpp
@@ -0,0 +1,137 @@
+#include <AggregateFunctions/AggregateFunctionUniqApacheHLL.h>
+#include <AggregateFunctions/AggregateFunctionFactory.h>
+#include <AggregateFunctions/FactoryHelpers.h>
+#include <Common/FieldVisitorConvertToNumber.h>
+
+#if USE_DATASKETCHES
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int ARGUMENT_OUT_OF_BOUND;
+    extern const int BAD_ARGUMENTS;
+    extern const int ILLEGAL_TYPE_OF_ARGUMENT;
+    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
+}
+
+static AggregateFunctionPtr createAggregateFunctionUniqApacheHLL(
+    const std::string & name, const DataTypes & argument_types, const Array & params, const Settings *)
+{
+    uint8_t lg_config_k = 12;
+    datasketches::target_hll_type target_type = datasketches::HLL_4;
+
+    if (params.size() > 2)
+        throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
+            "Aggregate function {} accepts at most two parameters (lg_k, type).", name);
+
+    if (!params.empty())
+    {
+        const UInt64 lg_k_param = applyVisitor(FieldVisitorConvertToNumber<UInt64>(), params[0]);
+        if (lg_k_param < 4 || lg_k_param > 21)
+            throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND,
+                "Parameter lg_k for aggregate function {} is out of range: [4, 21].", name);
+        lg_config_k = static_cast<uint8_t>(lg_k_param);
+    }
+
+    if (params.size() == 2)
+    {
+        if (params[1].getType() != Field::Types::String)
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Parameter type for aggregate function {} must be a string.", name);
+
+        const String type_param = params[1].safeGet<String>();
+        if (type_param == "HLL_4")
+            target_type = datasketches::HLL_4;
+        else if (type_param == "HLL_6")
+            target_type = datasketches::HLL_6;
+        else if (type_param == "HLL_8")
+            target_type = datasketches::HLL_8;
+        else
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Parameter type for aggregate function {} must be one of 'HLL_4', 'HLL_6', 'HLL_8'.", name);
+    }
+
+    assertUnary(name, argument_types);
+
+    const IDataType & argument_type = *argument_types[0];
+    WhichDataType which(argument_type);
+
+    /// Unlike other decimals, `DateTime64` has an interoperable representation as epoch ticks.
+    if (which.isDateTime64())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDateTime64::FieldType>>(lg_config_k, target_type, argument_types, params);
+
+    /// Exclude wide integers: DataSketches has no portable representation for them.
+    if (!which.isInt128() && !which.isInt256() && !which.isUInt128() && !which.isUInt256())
+    {
+        AggregateFunctionPtr res(createWithNumericType<AggregateFunctionUniqApacheHLL>(
+            argument_type, lg_config_k, target_type, argument_types, params));
+        if (res)
+            return res;
+    }
+
+    if (which.isDate())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDate::FieldType>>(lg_config_k, target_type, argument_types, params);
+    if (which.isDate32())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDate32::FieldType>>(lg_config_k, target_type, argument_types, params);
+    if (which.isDateTime())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDateTime::FieldType>>(lg_config_k, target_type, argument_types, params);
+    if (which.isStringOrFixedString())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<String>>(lg_config_k, target_type, argument_types, params);
+    if (which.isUUID())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeUUID::FieldType>>(lg_config_k, target_type, argument_types, params);
+    if (which.isIPv4())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv4::FieldType>>(lg_config_k, target_type, argument_types, params);
+    if (which.isIPv6())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv6::FieldType>>(lg_config_k, target_type, argument_types, params);
+
+    /// For `Nullable(Nothing)` the `Null` combinator replaces this function with `nothing`, but it must be created first.
+    if (argument_type.onlyNull())
+        return std::make_shared<AggregateFunctionUniqApacheHLL<UInt8>>(lg_config_k, target_type, argument_types, params);
+
+    throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT,
+        "Illegal type {} of argument for aggregate function {}. Use uniq, uniqCombined or uniqHLL12 for unsupported types.",
+        argument_type.getName(), name);
+}
+
+void registerAggregateFunctionUniqApacheHLL(AggregateFunctionFactory & factory);
+void registerAggregateFunctionUniqApacheHLL(AggregateFunctionFactory & factory)
+{
+    FunctionDocumentation::Description description = R"(
+Calculates the approximate number of different argument values using an [Apache DataSketches](https://datasketches.apache.org/docs/HLL/HllSketches.html) HyperLogLog sketch.
+
+The `-State` and `-Merge` combinators exchange Apache DataSketches HLL sketches with a varint length prefix.
+
+For interoperability, integers of at most 64 bits are hashed as 8-byte integers, floating-point values as IEEE-754 doubles, strings as raw bytes, `UUID` values as canonical 16 bytes, and `IPv6` addresses in network order.
+`Date`, `Date32`, `DateTime` and `DateTime64` use their underlying integer values. External producers must use the same unit; for example, `DateTime64(3)` uses epoch milliseconds.
+Unsupported types and multiple arguments are rejected. Use `uniq`, `uniqCombined` or `uniqHLL12` for those inputs.
+
+Merging can switch from the HIP estimator to the less accurate composite estimator, so results can depend on partitioning across threads, parts and shards.
+Merging a lower-resolution sketch permanently lowers the result's resolution, regardless of the declared `lg_k`.
+    )";
+    FunctionDocumentation::Syntax syntax = "uniqApacheHLL([lg_k, [type]])(x)";
+    FunctionDocumentation::Arguments arguments = {
+        {"x", "Column to compute the number of distinct values of.", {"(U)Int8/16/32/64", "Enum", "BFloat16", "Float32", "Float64", "String", "FixedString", "UUID", "IPv4", "IPv6", "Date", "Date32", "DateTime", "DateTime64"}},
+    };
+    FunctionDocumentation::Parameters parameters = {
+        {"lg_k", "Optional. Log-base-2 of the number of buckets, in range [4, 21]. Higher means better accuracy and more memory. Default: 12.", {"UInt8"}},
+        {"type", "Optional. Storage format of the sketch: 'HLL_4', 'HLL_6', or 'HLL_8'. Default: 'HLL_4'.", {"String"}},
+    };
+    FunctionDocumentation::ReturnedValue returned_value = {"Returns the approximate number of distinct values.", {"UInt64"}};
+    FunctionDocumentation::Examples examples = {
+        {"Basic usage", "SELECT uniqApacheHLL(number) FROM numbers(1000)", "1000"},
+        {"With parameters", "SELECT uniqApacheHLL(14, 'HLL_8')(number) FROM numbers(1000)", "1000"},
+    };
+    FunctionDocumentation::IntroducedIn introduced_in = {26, 6};
+    FunctionDocumentation::Category category = FunctionDocumentation::Category::AggregateFunction;
+    FunctionDocumentation documentation = {description, syntax, arguments, parameters, returned_value, examples, introduced_in, category};
+
+    AggregateFunctionProperties properties = { .returns_default_when_only_null = true, .is_order_dependent = false };
+
+    factory.registerFunction("uniqApacheHLL", {createAggregateFunctionUniqApacheHLL, documentation, properties});
+}
+
+}
+
+#endif
diff --git a/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.h b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.h
new file mode 100644
index 000000000000..a6325e9f0ca8
--- /dev/null
+++ b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.h
@@ -0,0 +1,243 @@
+#pragma once
+
+#include "config.h"
+
+#if USE_DATASKETCHES
+
+#include <AggregateFunctions/Helpers.h>
+#include <AggregateFunctions/IAggregateFunction.h>
+#include <Columns/ColumnDecimal.h>
+#include <Columns/ColumnsNumber.h>
+#include <Common/Exception.h>
+#include <Common/assert_cast.h>
+#include <Core/Field.h>
+#include <Core/UUID.h>
+#include <DataTypes/DataTypeDate.h>
+#include <DataTypes/DataTypeDate32.h>
+#include <DataTypes/DataTypeDateTime.h>
+#include <DataTypes/DataTypeDateTime64.h>
+#include <DataTypes/DataTypeIPv4andIPv6.h>
+#include <DataTypes/DataTypeUUID.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <IO/ReadBuffer.h>
+#include <IO/ReadHelpers.h>
+#include <IO/WriteBuffer.h>
+#include <IO/WriteHelpers.h>
+
+#include <hll.hpp>
+
+#include <cmath>
+#include <memory>
+#include <type_traits>
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int CORRUPTED_DATA;
+}
+
+
+/// Keeps insertion and union states separate to preserve the estimator used before merging.
+class HllSketchData
+{
+private:
+    std::unique_ptr<datasketches::hll_sketch> sk_update;
+    std::unique_ptr<datasketches::hll_union> sk_union;
+
+    datasketches::hll_sketch * getSkUpdate(uint8_t lg_config_k, datasketches::target_hll_type tgt_type)
+    {
+        if (!sk_update)
+            sk_update = std::make_unique<datasketches::hll_sketch>(lg_config_k, tgt_type);
+        return sk_update.get();
+    }
+
+    datasketches::hll_union * getSkUnion(uint8_t lg_config_k)
+    {
+        if (!sk_union)
+            sk_union = std::make_unique<datasketches::hll_union>(lg_config_k);
+        return sk_union.get();
+    }
+
+    void foldUpdateIntoUnionIfNeeded()
+    {
+        if (sk_union && sk_update)
+        {
+            sk_union->update(*sk_update);
+            sk_update.reset();
+        }
+    }
+
+public:
+    template <typename T>
+    void insert(T value, uint8_t lg_config_k, datasketches::target_hll_type tgt_type)
+    {
+        getSkUpdate(lg_config_k, tgt_type)->update(value);
+        foldUpdateIntoUnionIfNeeded();
+    }
+
+    void insertData(const char * data, size_t size, uint8_t lg_config_k, datasketches::target_hll_type tgt_type)
+    {
+        getSkUpdate(lg_config_k, tgt_type)->update(static_cast<const void *>(data), size);
+        foldUpdateIntoUnionIfNeeded();
+    }
+
+    UInt64 size(datasketches::target_hll_type tgt_type) const
+    {
+        /// Rounding preserves exact cardinalities despite floating-point error.
+        if (sk_union)
+            return static_cast<UInt64>(std::llround(sk_union->get_result(tgt_type).get_estimate()));
+        if (sk_update)
+            return static_cast<UInt64>(std::llround(sk_update->get_estimate()));
+        return 0;
+    }
+
+    void merge(const HllSketchData & rhs, uint8_t lg_config_k, datasketches::target_hll_type tgt_type)
+    {
+        datasketches::hll_union * u = getSkUnion(lg_config_k);
+
+        foldUpdateIntoUnionIfNeeded();
+
+        if (rhs.sk_update)
+            u->update(*rhs.sk_update);
+        else if (rhs.sk_union)
+            u->update(rhs.sk_union->get_result(tgt_type));
+    }
+
+    /// You can only call this for an empty object.
+    void read(ReadBuffer & in, uint8_t lg_config_k)
+    {
+        datasketches::hll_sketch::vector_bytes bytes;
+        readVectorBinary(bytes, in);
+        if (bytes.empty())
+            return;
+
+        try
+        {
+            auto sk = datasketches::hll_sketch::deserialize(bytes.data(), bytes.size());
+            getSkUnion(lg_config_k)->update(std::move(sk));
+        }
+        catch (const DB::Exception &)
+        {
+            throw;
+        }
+        catch (const std::bad_alloc &)
+        {
+            throw;
+        }
+        catch (const std::exception & e)
+        {
+            /// Translate malformed input to avoid a logical exception in `SerializationAggregateFunction`.
+            throw Exception(ErrorCodes::CORRUPTED_DATA, "Cannot deserialize HLL sketch state: {}", e.what());
+        }
+    }
+
+    void write(WriteBuffer & out, uint8_t lg_config_k, datasketches::target_hll_type tgt_type) const
+    {
+        datasketches::hll_sketch::vector_bytes bytes;
+        if (sk_update)
+            bytes = sk_update->serialize_compact();
+        else if (sk_union)
+            bytes = sk_union->get_result(tgt_type).serialize_compact();
+        else
+            bytes = datasketches::hll_sketch(lg_config_k, tgt_type).serialize_compact();
+        writeVectorBinary(bytes, out);
+    }
+};
+
+
+template <typename T>
+class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper<HllSketchData, AggregateFunctionUniqApacheHLL<T>>
+{
+    using Base = IAggregateFunctionDataHelper<HllSketchData, AggregateFunctionUniqApacheHLL<T>>;
+
+    uint8_t lg_config_k;
+    datasketches::target_hll_type target_type;
+
+public:
+    AggregateFunctionUniqApacheHLL(
+        uint8_t lg_config_k_,
+        datasketches::target_hll_type target_type_,
+        const DataTypes & argument_types_,
+        const Array & params_)
+        : Base(argument_types_, params_, std::make_shared<DataTypeUInt64>())
+        , lg_config_k(lg_config_k_)
+        , target_type(target_type_)
+    {
+    }
+
+    String getName() const override { return "uniqApacheHLL"; }
+
+    bool allocatesMemoryInArena() const override { return false; }
+
+    void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override
+    {
+        auto & data = this->data(place);
+
+        if constexpr (std::is_same_v<T, String>)
+        {
+            const auto value = columns[0]->getDataAt(row_num);
+            data.insertData(value.data(), value.size(), lg_config_k, target_type);
+        }
+        else
+        {
+            const auto & value = assert_cast<const ColumnVectorOrDecimal<T> &>(*columns[0]).getData()[row_num];
+
+            if constexpr (std::is_same_v<T, UUID>)
+            {
+                /// Convert the two host-order halves of a `UUID` to canonical bytes.
+                const UInt64 halves[2] = {
+                    std::byteswap(UUIDHelpers::getHighBytes(value)),
+                    std::byteswap(UUIDHelpers::getLowBytes(value)),
+                };
+                data.insertData(reinterpret_cast<const char *>(halves), sizeof(halves), lg_config_k, target_type);
+            }
+            else if constexpr (std::is_same_v<T, IPv6>)
+                /// Already held in network order, which is the canonical form.
+                data.insertData(reinterpret_cast<const char *>(&value), sizeof(value), lg_config_k, target_type);
+            else if constexpr (is_decimal<T>)
+                /// Hash `DateTime64` as epoch ticks; external producers must use the same scale.
+                data.insert(static_cast<Int64>(value.value), lg_config_k, target_type);
+            else if constexpr (std::is_same_v<T, IPv4>)
+                data.insert(static_cast<UInt64>(value.toUnderType()), lg_config_k, target_type);
+            else if constexpr (std::is_same_v<T, BFloat16> || std::is_floating_point_v<T>)
+                data.insert(static_cast<Float64>(value), lg_config_k, target_type);
+            else if constexpr (std::is_signed_v<T>)
+                data.insert(static_cast<Int64>(value), lg_config_k, target_type);
+            else
+                data.insert(static_cast<UInt64>(value), lg_config_k, target_type);
+        }
+    }
+
+    /// Serialized sketches carry their configuration, so parameters need not match.
+    bool haveSameStateRepresentationImpl(const IAggregateFunction & rhs) const override
+    {
+        return getName() == rhs.getName() && this->haveEqualArgumentTypes(rhs);
+    }
+
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override
+    {
+        this->data(place).merge(this->data(rhs), lg_config_k, target_type);
+    }
+
+    void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
+    {
+        this->data(place).write(buf, lg_config_k, target_type);
+    }
+
+    void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
+    {
+        this->data(place).read(buf, lg_config_k);
+    }
+
+    void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
+    {
+        assert_cast<ColumnUInt64 &>(to).getData().push_back(this->data(place).size(target_type));
+    }
+
+};
+
+}
+
+#endif
diff --git a/src/AggregateFunctions/registerAggregateFunctions.cpp b/src/AggregateFunctions/registerAggregateFunctions.cpp
index 46997750a403..670180d87a46 100644
--- a/src/AggregateFunctions/registerAggregateFunctions.cpp
+++ b/src/AggregateFunctions/registerAggregateFunctions.cpp
@@ -61,6 +61,9 @@ void registerAggregateFunctionSumMap(AggregateFunctionFactory &);
 void registerAggregateFunctionsUniq(AggregateFunctionFactory &);
 void registerAggregateFunctionUniqCombined(AggregateFunctionFactory &);
 void registerAggregateFunctionUniqUpTo(AggregateFunctionFactory &);
+#if USE_DATASKETCHES
+void registerAggregateFunctionUniqApacheHLL(AggregateFunctionFactory &);
+#endif
 void registerAggregateFunctionTopK(AggregateFunctionFactory &);
 void registerAggregateFunctionsBitwise(AggregateFunctionFactory &);
 void registerAggregateFunctionsBitmap(AggregateFunctionFactory &);
@@ -174,6 +177,9 @@ void registerAggregateFunctions()
         registerAggregateFunctionsUniq(factory);
         registerAggregateFunctionUniqCombined(factory);
         registerAggregateFunctionUniqUpTo(factory);
+#if USE_DATASKETCHES
+        registerAggregateFunctionUniqApacheHLL(factory);
+#endif
         registerAggregateFunctionTopK(factory);
         registerAggregateFunctionsBitwise(factory);
         registerAggregateFunctionCramersV(factory);
diff --git a/tests/queries/0_stateless/04327_uniq_apache_hll.reference b/tests/queries/0_stateless/04327_uniq_apache_hll.reference
new file mode 100644
index 000000000000..3939bfc12e81
--- /dev/null
+++ b/tests/queries/0_stateless/04327_uniq_apache_hll.reference
@@ -0,0 +1,57 @@
+accuracy
+1
+1
+1	1
+empty and single
+0
+1
+state and merge
+1
+1
+AggregateFunction(uniqApacheHLL(14, \'HLL_8\'), UInt64)
+parameter validation
+argument types
+20
+20
+20
+20
+20
+20
+20
+20
+20
+20
+20
+3
+nullable and low cardinality
+20
+20
+0
+0
+0
+UInt64	0
+unsupported types
+one argument only
+import sketches built outside ClickHouse
+5
+11
+84
+5
+0
+types wider than the 8 bytes update(long) takes
+1
+1
+DateTime64 is hashed as the epoch time it counts
+1
+export sketches for consumption outside ClickHouse
+1
+1
+1
+store and merge states with a different lg_k
+20
+merging states built with a different lg_k
+1
+1
+relabelling with CAST does not rescale
+1	1
+only the parameters are interchangeable
diff --git a/tests/queries/0_stateless/04327_uniq_apache_hll.sql b/tests/queries/0_stateless/04327_uniq_apache_hll.sql
new file mode 100644
index 000000000000..4d2a4bf5d477
--- /dev/null
+++ b/tests/queries/0_stateless/04327_uniq_apache_hll.sql
@@ -0,0 +1,162 @@
+-- Tags: no-fasttest
+-- ^ DataSketches is not built in fast-test builds.
+
+SELECT 'accuracy';
+SELECT abs(toInt64(uniqApacheHLL(number)) - 100000) < 3000 FROM numbers(100000);
+SELECT abs(toInt64(uniqApacheHLL(14)(number)) - 100000) < 1500 FROM numbers(100000);
+-- Storage type does not change the estimate.
+SELECT
+    uniqApacheHLL(12, 'HLL_4')(number) = uniqApacheHLL(12, 'HLL_6')(number),
+    uniqApacheHLL(12, 'HLL_4')(number) = uniqApacheHLL(12, 'HLL_8')(number)
+FROM numbers(1000);
+
+SELECT 'empty and single';
+SELECT uniqApacheHLL(number) FROM numbers(0);
+SELECT uniqApacheHLL(number) FROM numbers(1);
+
+SELECT 'state and merge';
+-- Direct and merged sketches may use different estimators.
+SELECT
+    abs(toInt64(uniqApacheHLLMerge(s)) - toInt64((SELECT uniqApacheHLL(number) FROM numbers(100000)))) < 3000
+FROM
+(
+    SELECT uniqApacheHLLState(number) AS s
+    FROM numbers(100000)
+    GROUP BY number % 17
+);
+
+-- Different partitions must produce the same union estimate.
+SELECT
+    (SELECT uniqApacheHLLMerge(s) FROM (SELECT uniqApacheHLLState(number) AS s FROM numbers(100000) GROUP BY number % 17))
+  = (SELECT uniqApacheHLLMerge(s) FROM (SELECT uniqApacheHLLState(number) AS s FROM numbers(100000) GROUP BY number % 13));
+
+SELECT toTypeName(uniqApacheHLLState(14, 'HLL_8')(number)) FROM numbers(1);
+
+SELECT 'parameter validation';
+SELECT uniqApacheHLL(3)(number) FROM numbers(1); -- { serverError ARGUMENT_OUT_OF_BOUND }
+SELECT uniqApacheHLL(22)(number) FROM numbers(1); -- { serverError ARGUMENT_OUT_OF_BOUND }
+SELECT uniqApacheHLL(12, 'HLL_9')(number) FROM numbers(1); -- { serverError BAD_ARGUMENTS }
+SELECT uniqApacheHLL(12, 'HLL_4', 1)(number) FROM numbers(1); -- { serverError NUMBER_OF_ARGUMENTS_DOESNT_MATCH }
+
+SELECT 'argument types';
+-- Small counts stay in coupon mode, where the estimate is exact.
+SELECT uniqApacheHLL(toUInt64(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toInt32(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toBFloat16(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toFloat32(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toFloat64(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toString(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toFixedString(toString(number), 8)) FROM numbers(20);
+SELECT uniqApacheHLL(toDate('2020-01-01') + number) FROM numbers(20);
+SELECT uniqApacheHLL(toDate32('2020-01-01') + number) FROM numbers(20);
+SELECT uniqApacheHLL(toDateTime('2020-01-01 00:00:00') + number) FROM numbers(20);
+SELECT uniqApacheHLL(toIPv4('1.2.3.0') + number) FROM numbers(20);
+SELECT uniqApacheHLL(CAST(number % 3, 'Enum8(\'a\' = 0, \'b\' = 1, \'c\' = 2)')) FROM numbers(20);
+
+SELECT 'nullable and low cardinality';
+SELECT uniqApacheHLL(toNullable(number)) FROM numbers(20);
+SELECT uniqApacheHLL(toLowCardinality(toString(number))) FROM numbers(20);
+-- All-NULL argument returns 0, as `uniq` does.
+SELECT uniqApacheHLL(NULL);
+SELECT uniqApacheHLL(14, 'HLL_8')(NULL);
+SELECT uniqApacheHLL(NULL) FROM numbers(10);
+SELECT toTypeName(uniqApacheHLLState(NULL)), uniqApacheHLLState(NULL);
+
+SELECT 'unsupported types';
+SELECT uniqApacheHLL(toInt128(number)) FROM numbers(20); -- { serverError ILLEGAL_TYPE_OF_ARGUMENT }
+SELECT uniqApacheHLL(toUInt256(number)) FROM numbers(20); -- { serverError ILLEGAL_TYPE_OF_ARGUMENT }
+SELECT uniqApacheHLL(toDecimal64(number, 2)) FROM numbers(20); -- { serverError ILLEGAL_TYPE_OF_ARGUMENT }
+SELECT uniqApacheHLL(materialize([number])) FROM numbers(20); -- { serverError ILLEGAL_TYPE_OF_ARGUMENT }
+SELECT uniqApacheHLL((number, number + 1)) FROM numbers(20); -- { serverError ILLEGAL_TYPE_OF_ARGUMENT }
+
+SELECT 'one argument only';
+SELECT uniqApacheHLL(number, number + 1) FROM numbers(20); -- { serverError NUMBER_OF_ARGUMENTS_DOESNT_MATCH }
+SELECT uniqApacheHLL() FROM numbers(1); -- { serverError NUMBER_OF_ARGUMENTS_DOESNT_MATCH }
+
+-- Fixtures were generated by Apache DataSketches C++, with a varint length prefix.
+-- These checks pin compatibility with external producers and consumers.
+
+SELECT 'import sketches built outside ClickHouse';
+
+-- lg_k = 12, HLL_4, `hll_sketch::update` on `uint64_t` values over 0..4. Coupon list mode.
+SELECT finalizeAggregation(CAST(unhex('1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06'), 'AggregateFunction(uniqApacheHLL, UInt64)'));
+
+-- External sketches over 0..4 and 3..7 overlap with ClickHouse values 6..10.
+SELECT uniqApacheHLLMerge(s) FROM
+(
+    SELECT CAST(unhex('1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06'), 'AggregateFunction(uniqApacheHLL, UInt64)') AS s
+    UNION ALL
+    SELECT CAST(unhex('1C0201070C030805007581660781BC5D067B65E608FC2D420AC1E91705'), 'AggregateFunction(uniqApacheHLL, UInt64)') AS s
+    UNION ALL
+    SELECT uniqApacheHLLState(number) AS s FROM numbers(6, 5)
+);
+
+-- lg_k = 4, HLL_4, 100 distinct values. Dense HLL mode rather than a coupon list.
+SELECT finalizeAggregation(CAST(unhex('300A0107040008020215EB1DC787F15440000000000000FB3F000000000000000003000000000000000251214121031025'), 'AggregateFunction(uniqApacheHLL(4), UInt64)'));
+
+-- lg_k = 14, HLL_8: the non-default parameters are carried by the state type.
+SELECT finalizeAggregation(CAST(unhex('1C0201070E03080508CBD7C2042BF2FB06862FF90D7581660781BC5D06'), 'AggregateFunction(uniqApacheHLL(14, \'HLL_8\'), UInt64)'));
+
+-- An externally produced empty sketch.
+SELECT finalizeAggregation(CAST(unhex('080201070C030C0000'), 'AggregateFunction(uniqApacheHLL, UInt64)'));
+
+SELECT 'types wider than the 8 bytes update(long) takes';
+
+-- Canonical bytes of `UUID` values 00000000-0000-0000-0000-00000000000{0..4}.
+SELECT hex(toString(uniqApacheHLLState(toUUID(concat('00000000-0000-0000-0000-00000000000', toString(number))))))
+     = '1C0201070C0308050050C94D05854BD10ADB8CBD053C56FB07F8FDB206'
+FROM numbers(5) SETTINGS max_threads = 1;
+
+-- The same for 2001:db8::1 .. ::5, which ClickHouse already holds in network order.
+SELECT hex(toString(uniqApacheHLLState(toIPv6(concat('2001:db8::', hex(number + 1))))))
+     = '1C0201070C0308050018216E09FAB4750F52D5BB07DFBDE30A79BC9D0B'
+FROM numbers(5) SETTINGS max_threads = 1;
+
+SELECT 'DateTime64 is hashed as the epoch time it counts';
+
+-- Epoch milliseconds for 2020-01-01 00:00:00.000 UTC and the next four seconds.
+SELECT hex(toString(uniqApacheHLLState(toDateTime64('2020-01-01 00:00:00.000', 3, 'UTC') + number)))
+     = '1C0201070C0308050001D4B019CBDD6F1059310D0E833938083897B304'
+FROM numbers(5) SETTINGS max_threads = 1;
+
+SELECT 'export sketches for consumption outside ClickHouse';
+
+-- Use one thread to keep coupon ordering deterministic.
+SELECT hex(toString(uniqApacheHLLState(number))) = '1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06' FROM numbers(5) SETTINGS max_threads = 1;
+SELECT hex(toString(uniqApacheHLLState(14, 'HLL_8')(number))) = '1C0201070E03080508CBD7C2042BF2FB06862FF90D7581660781BC5D06' FROM numbers(5) SETTINGS max_threads = 1;
+
+-- Importing an external sketch and exporting it again must reproduce it byte for byte.
+SELECT hex(toString(uniqApacheHLLMergeState(s))) = '1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06'
+FROM (SELECT CAST(unhex('1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06'), 'AggregateFunction(uniqApacheHLL, UInt64)') AS s)
+SETTINGS max_threads = 1;
+
+SELECT 'store and merge states with a different lg_k';
+
+DROP TABLE IF EXISTS hll_interop_states;
+CREATE TABLE hll_interop_states (k UInt8, s AggregateFunction(uniqApacheHLL, UInt64)) ENGINE = AggregatingMergeTree ORDER BY k;
+INSERT INTO hll_interop_states SELECT number % 4 AS k, uniqApacheHLLState(8)(number) FROM numbers(20) GROUP BY k;
+OPTIMIZE TABLE hll_interop_states FINAL;
+-- Small enough to stay in coupon mode, so the union is exact regardless of how it was partitioned.
+SELECT uniqApacheHLLMerge(s) FROM hll_interop_states;
+DROP TABLE hll_interop_states;
+
+SELECT 'merging states built with a different lg_k';
+-- Downsampling and building at the lower resolution must produce the same union estimate.
+SELECT
+    (SELECT uniqApacheHLLMerge(8)(s) FROM (SELECT uniqApacheHLLState(number)    AS s FROM numbers(100000) GROUP BY number % 17))
+  = (SELECT uniqApacheHLLMerge(8)(s) FROM (SELECT uniqApacheHLLState(8)(number) AS s FROM numbers(100000) GROUP BY number % 17));
+
+-- Downsampling must also reduce the serialized state size.
+SELECT
+    length(toString(uniqApacheHLLMergeState(8)(s))) < length(toString(uniqApacheHLLMergeState(s))) / 8
+FROM (SELECT uniqApacheHLLState(number) AS s FROM numbers(100000) GROUP BY number % 17);
+
+SELECT 'relabelling with CAST does not rescale';
+SELECT
+    finalizeAggregation(CAST(s, 'AggregateFunction(uniqApacheHLL(8), UInt64)')) = finalizeAggregation(s),
+    length(toString(CAST(s, 'AggregateFunction(uniqApacheHLL(8), UInt64)'))) = length(toString(s))
+FROM (SELECT uniqApacheHLLState(number) AS s FROM numbers(1000));
+
+SELECT 'only the parameters are interchangeable';
+SELECT CAST(uniqApacheHLLState(toString(number)), 'AggregateFunction(uniqApacheHLL, UInt64)') FROM numbers(10); -- { serverError CANNOT_CONVERT_TYPE }
+SELECT CAST(uniqThetaState(number), 'AggregateFunction(uniqApacheHLL, UInt64)') FROM numbers(10); -- { serverError CANNOT_CONVERT_TYPE }
diff --git a/tests/queries/0_stateless/05055_uniq_theta_compressed_state.reference b/tests/queries/0_stateless/05055_uniq_theta_compressed_state.reference
new file mode 100644
index 000000000000..5a931d5663aa
--- /dev/null
+++ b/tests/queries/0_stateless/05055_uniq_theta_compressed_state.reference
@@ -0,0 +1 @@
+16	16	30
diff --git a/tests/queries/0_stateless/05055_uniq_theta_compressed_state.sql b/tests/queries/0_stateless/05055_uniq_theta_compressed_state.sql
new file mode 100644
index 000000000000..848cfc356e9f
--- /dev/null
+++ b/tests/queries/0_stateless/05055_uniq_theta_compressed_state.sql
@@ -0,0 +1,14 @@
+-- Tags: no-fasttest
+-- - no-fasttest -- compiled w/o datasketches
+
+-- Compressed Theta states (serialization version 4) exercise separate 33- and 35-bit
+-- unpacking routines that previously underestimated cardinality. Each retains 16 values
+-- with theta = 1; two shared values give a union of 30.
+
+WITH
+    CAST(unhex('4B01040321011ACC9310000001F4000000FA0000007D0000003E8000001F4000000FA0000007D180000000000001F4000000FA0000007D0000003E8000001F4000000FA0000007D0000003E8') AS AggregateFunction(uniqTheta, UInt64)) AS state_33_bits,
+    CAST(unhex('4F01040323011ACC93100000007D0000000FA1000000004000000000000007D0000000FA0000001F40000003E80000007D0000000FA1000000000000003E80000007D0000000FA0000001F40000003E8') AS AggregateFunction(uniqTheta, UInt64)) AS state_35_bits
+SELECT
+    finalizeAggregation(state_33_bits),
+    finalizeAggregation(state_35_bits),
+    finalizeAggregation(uniqThetaUnion(state_33_bits, state_35_bits));
diff --git a/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.reference b/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.reference
new file mode 100644
index 000000000000..d7c73d3f18be
--- /dev/null
+++ b/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.reference
@@ -0,0 +1,2 @@
+OK unknown type
+OK rowbinary
diff --git a/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.sh b/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.sh
new file mode 100755
index 000000000000..80c16c7e17f6
--- /dev/null
+++ b/tests/queries/0_stateless/05136_uniq_apache_hll_corrupted_state.sh
@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+# Tags: no-fasttest
+# no-fasttest -- compiled w/o datasketches
+
+# Malformed states must raise `CORRUPTED_DATA`, not a logical exception.
+# A shell test tolerates the client's extra stack trace, as in
+# `04307_uniqTheta_corrupted_state_106259.sh`.
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+# Valid length prefix, invalid sketch preamble.
+$CLICKHOUSE_CLIENT --query \
+    "SELECT finalizeAggregation(CAST(unhex('08FFFFFFFFFFFFFFFF'), 'AggregateFunction(uniqApacheHLL, UInt64)'))" 2>&1 \
+    | grep -q -F 'CORRUPTED_DATA' && echo 'OK unknown type' || echo 'FAIL unknown type'
+
+# The `RowBinary` payload is shorter than any valid HLL sketch.
+printf '\x03\x03\x03\x30\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' \
+    | $CLICKHOUSE_LOCAL --input-format=RowBinary \
+        --structure='x AggregateFunction(uniqApacheHLL, IPv6)' \
+        --query='SELECT x FROM table' 2>&1 \
+    | grep -q -F 'CORRUPTED_DATA' && echo 'OK rowbinary' || echo 'FAIL rowbinary'