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..4e91777839ba 160000
--- a/contrib/datasketches-cpp
+++ b/contrib/datasketches-cpp
@@ -1 +1 @@
-Subproject commit 76edd74f5db286b672c170a8ded4ce39b3a8800f
+Subproject commit 4e91777839ba01c43f0da7d049f15478301726d3
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..95144e70c22a
--- /dev/null
+++ b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.cpp
@@ -0,0 +1,140 @@
+#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.
+
+`NULL` values and empty strings are ignored, as in the Java, Python and C++ implementations and in Spark; unlike `uniq`, an empty string is not counted as a value.
+A `Nullable` argument gives the same state as a non-`Nullable` one, a bare DataSketches sketch, and the same state type, `AggregateFunction(uniqApacheHLL, T)`.
+
+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..76a0cd41de68
--- /dev/null
+++ b/src/AggregateFunctions/AggregateFunctionUniqApacheHLL.h
@@ -0,0 +1,275 @@
+#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/DataTypeNullable.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 <algorithm>
+#include <bit>
+#include <cmath>
+#include <exception>
+#include <memory>
+#include <new>
+#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:
+    /// Folding is deferred until the state is read, so it can happen in const methods.
+    mutable std::unique_ptr<datasketches::hll_sketch> sk_update;
+    mutable 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();
+    }
+
+    /// Inserts keep going into `sk_update` even when `sk_union` exists, to avoid creating and merging a sketch per row.
+    void foldUpdateIntoUnionIfNeeded() const
+    {
+        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);
+    }
+
+    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);
+    }
+
+    UInt64 size() const
+    {
+        foldUpdateIntoUnionIfNeeded();
+
+        /// Rounding preserves exact cardinalities despite floating-point error.
+        if (sk_union)
+            return static_cast<UInt64>(std::llround(sk_union->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)
+    {
+        if (!rhs.sk_update && !rhs.sk_union)
+            return;
+
+        datasketches::hll_union * u = getSkUnion(lg_config_k);
+
+        /// `rhs` may hold both sketches; take both without modifying it.
+        if (rhs.sk_update)
+            u->update(*rhs.sk_update);
+
+        /// The union accepts a sketch of any type, and `HLL_8` is its own type, so there is no re-encoding.
+        if (rhs.sk_union)
+            u->update(rhs.sk_union->get_result(datasketches::HLL_8));
+
+        foldUpdateIntoUnionIfNeeded();
+    }
+
+    /// 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
+    {
+        foldUpdateIntoUnionIfNeeded();
+
+        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; }
+
+    /// A `Nullable` argument only has its `NULL` rows skipped: the state is the same bare sketch, with the same type.
+    bool stateIsIndependentOfNullability() const override { return true; }
+
+    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);
+
+            /// Other DataSketches implementations ignore empty strings, and the sketch of a mixed
+            /// input must be the same as theirs.
+            if (value.size() == 0)
+                return;
+
+            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, and nullability of the argument does not matter.
+    bool haveSameStateRepresentationImpl(const IAggregateFunction & rhs) const override
+    {
+        const auto & lhs_types = this->getArgumentTypes();
+        const auto & rhs_types = rhs.getArgumentTypes();
+
+        return getName() == rhs.getName()
+            && std::equal(
+                lhs_types.begin(),
+                lhs_types.end(),
+                rhs_types.begin(),
+                rhs_types.end(),
+                [](const auto & lhs, const auto & rhs_type) { return removeNullable(lhs)->equals(*removeNullable(rhs_type)); });
+    }
+
+    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override
+    {
+        this->data(place).merge(this->data(rhs), lg_config_k);
+    }
+
+    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());
+    }
+};
+
+}
+
+#endif
diff --git a/src/AggregateFunctions/Combinators/AggregateFunctionIf.cpp b/src/AggregateFunctions/Combinators/AggregateFunctionIf.cpp
index 1818e9499699..2d630965076b 100644
--- a/src/AggregateFunctions/Combinators/AggregateFunctionIf.cpp
+++ b/src/AggregateFunctions/Combinators/AggregateFunctionIf.cpp
@@ -113,6 +113,19 @@ class AggregateFunctionIfNullUnary final
         filter_is_only_null = arguments[num_arguments - 1]->onlyNull();
     }
 
+protected:
+    /// The nested function is the one without `If`, so the state type of the `If` function without Nullable arguments is built here.
+    DataTypePtr getNestedStateType() const override
+    {
+        DataTypes plain_argument_types;
+        plain_argument_types.reserve(this->getArgumentTypes().size());
+        for (const auto & type : this->getArgumentTypes())
+            plain_argument_types.push_back(type->onlyNull() ? type : removeNullable(type));
+
+        return std::make_shared<AggregateFunctionIf>(this->nested_function, plain_argument_types, this->getParameters())->getStateType();
+    }
+
+public:
     void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena * arena) const override
     {
         if (filter_is_only_null)
@@ -488,6 +501,20 @@ AggregateFunctionPtr AggregateFunctionIf::getOwnNullAdapter(
 
     bool need_to_serialize_flag = return_type_is_nullable || properties.returns_default_when_only_null;
 
+    /// The function only skips NULL rows: no flag, and the state of the function without Nullable arguments, with its type.
+    if (nested_func->stateIsIndependentOfNullability())
+    {
+        if (!properties.returns_default_when_only_null)
+            throw Exception(ErrorCodes::LOGICAL_ERROR,
+                "Aggregate function {} cannot have the state independent of nullability unless it returns a default value for NULL values only",
+                nested_function->getName());
+
+        if (arguments.size() <= 2 && arguments.front()->isNullable())
+            return std::make_shared<AggregateFunctionIfNullUnary<false, false>>(nested_function->getName(), nested_func, arguments, params);
+
+        return std::make_shared<AggregateFunctionIfNullVariadic<false, false>>(nested_function, arguments, params);
+    }
+
     if (arguments.size() <= 2 && arguments.front()->isNullable())
     {
         if (return_type_is_nullable)
diff --git a/src/AggregateFunctions/Combinators/AggregateFunctionIf.h b/src/AggregateFunctions/Combinators/AggregateFunctionIf.h
index 4161b7937148..a325e10c355f 100644
--- a/src/AggregateFunctions/Combinators/AggregateFunctionIf.h
+++ b/src/AggregateFunctions/Combinators/AggregateFunctionIf.h
@@ -64,6 +64,11 @@ class AggregateFunctionIf final : public IAggregateFunctionHelper<AggregateFunct
         return nested_func->getNormalizedStateType();
     }
 
+    bool stateIsIndependentOfNullability() const override
+    {
+        return nested_func->stateIsIndependentOfNullability();
+    }
+
     bool canMergeStateFromDifferentVariant(const IAggregateFunction & rhs) const override
     {
         if (!this->haveSameDefinition(rhs))
diff --git a/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp b/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp
index fabc256c90da..f9cc1570c2bd 100644
--- a/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp
+++ b/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp
@@ -13,6 +13,7 @@ namespace DB
 namespace ErrorCodes
 {
     extern const int ILLEGAL_TYPE_OF_ARGUMENT;
+    extern const int LOGICAL_ERROR;
 }
 
 namespace
@@ -126,6 +127,19 @@ class AggregateFunctionCombinatorNull final : public IAggregateFunctionCombinato
             return new_function;
         }
 
+        /// The function only skips NULL rows: no flag, and the state of the nested function, with its type.
+        if (nested_function->stateIsIndependentOfNullability())
+        {
+            if (!properties.returns_default_when_only_null)
+                throw Exception(ErrorCodes::LOGICAL_ERROR,
+                    "Aggregate function {} cannot have the state independent of nullability unless it returns a default value for NULL values only",
+                    nested_function->getName());
+
+            if (arguments.size() == 1)
+                return std::make_shared<AggregateFunctionNullUnary<false, false>>(nested_function, arguments, params);
+            return std::make_shared<AggregateFunctionNullVariadic<false, false>>(nested_function, arguments, params);
+        }
+
         bool return_type_is_nullable = !properties.returns_default_when_only_null && nested_function->getResultType()->canBeInsideNullable();
         bool serialize_flag = return_type_is_nullable || properties.returns_default_when_only_null;
 
diff --git a/src/AggregateFunctions/Combinators/AggregateFunctionNull.h b/src/AggregateFunctions/Combinators/AggregateFunctionNull.h
index c18c68098a8e..f5ca11d7e93e 100644
--- a/src/AggregateFunctions/Combinators/AggregateFunctionNull.h
+++ b/src/AggregateFunctions/Combinators/AggregateFunctionNull.h
@@ -50,6 +50,19 @@ class AggregateFunctionNullBase : public IAggregateFunctionHelper<Derived>
     const AggregateFunctionPtr nested_function;
     const size_t prefix_size;
 
+    /// The type of the state of the nested function, which this wrapper has when the nested function
+    /// says its state does not depend on nullability. See `IAggregateFunction::stateIsIndependentOfNullability`.
+    virtual DataTypePtr getNestedStateType() const { return nested_function->getStateType(); }
+
+    /// Only a wrapper without the flag has the memory layout of the nested state, which the nested function can claim for its own type.
+    bool hasStateOfNestedFunction() const
+    {
+        if constexpr (!result_is_nullable && !serialize_flag)
+            return nested_function->stateIsIndependentOfNullability();
+        else
+            return false;
+    }
+
     /** In addition to data for nested aggregate function, we keep a flag
       *  indicating - was there at least one non-NULL value accumulated.
       * In case of no not-NULL values, the function will return NULL.
@@ -107,6 +120,20 @@ class AggregateFunctionNullBase : public IAggregateFunctionHelper<Derived>
         return nested_function->getName();
     }
 
+    DataTypePtr getStateType() const override
+    {
+        if (hasStateOfNestedFunction())
+            return getNestedStateType();
+        return IAggregateFunction::getStateType();
+    }
+
+    DataTypePtr getNormalizedStateType() const override
+    {
+        if (hasStateOfNestedFunction())
+            return nested_function->getNormalizedStateType();
+        return IAggregateFunction::getNormalizedStateType();
+    }
+
     bool canMergeStateFromDifferentVariant(const IAggregateFunction & rhs) const override
     {
         if (!this->haveSameDefinition(rhs))
diff --git a/src/AggregateFunctions/IAggregateFunction.h b/src/AggregateFunctions/IAggregateFunction.h
index b666df47c664..6f0cbacd7158 100644
--- a/src/AggregateFunctions/IAggregateFunction.h
+++ b/src/AggregateFunctions/IAggregateFunction.h
@@ -417,6 +417,19 @@ class IAggregateFunction : public std::enable_shared_from_this<IAggregateFunctio
     /// This method returns the set of argument indexes that can be always NULL, they will be skipped in combinator Null.
     virtual UnorderedSetWithMemoryTracking<size_t> getArgumentsThatCanBeOnlyNull() const { return {}; }
 
+    /** When the function is wrapped with the Null combinator (or the If combinator over Nullable arguments), the state of the
+      * wrapper has a flag byte and a type that names the Nullable argument. So the state of a Nullable column is not the state of a
+      * plain one: another implementation cannot read it, and the two cannot be mixed.
+      *
+      * A function that only skips NULL values, and whose state of no values is the empty state (example: uniqApacheHLL),
+      * does not need that. If it returns true here, the wrapper only skips NULL rows, and its state has the same bytes and the
+      * same type as the state of this function for a non-Nullable argument.
+      *
+      * It requires returns_default_when_only_null: the result of a function that must return NULL when only NULL values
+      * were seen needs the flag to tell it from the state of no rows.
+      */
+    virtual bool stateIsIndependentOfNullability() const { return false; }
+
     /** Return the nested function if this is an Aggregate Function Combinator.
       * Otherwise return nullptr.
       */
diff --git a/src/AggregateFunctions/registerAggregateFunctions.cpp b/src/AggregateFunctions/registerAggregateFunctions.cpp
index 46997750a403..22291d140905 100644
--- a/src/AggregateFunctions/registerAggregateFunctions.cpp
+++ b/src/AggregateFunctions/registerAggregateFunctions.cpp
@@ -1,5 +1,7 @@
 #include <AggregateFunctions/registerAggregateFunctions.h>
 
+#include "config.h"
+
 #include <AggregateFunctions/AggregateFunctionFactory.h>
 #include <AggregateFunctions/Combinators/AggregateFunctionCombinatorFactory.h>
 
@@ -61,6 +63,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 +179,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..cf1111734501
--- /dev/null
+++ b/tests/queries/0_stateless/04327_uniq_apache_hll.sql
@@ -0,0 +1,163 @@
+-- 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(-1)(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'
diff --git a/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.reference b/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.reference
new file mode 100644
index 000000000000..97f02e1a3b1a
--- /dev/null
+++ b/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.reference
@@ -0,0 +1,65 @@
+integers
+Int8	1C0201070C030805001FABB112D964E70BCBD7C2042BF2FB062F537613
+UInt8	180201070C03080400CBD7C2042F53761320A9190E34F7530A
+Int16	180201070C0308040026276604D964E70BCBD7C204A82F2604
+UInt16	180201070C03080400CBD7C204A82F26040066F004710BB407
+Int32	180201070C0308040047E5A708D964E70BCBD7C2043F53B008
+UInt32	180201070C03080400CBD7C2043F53B008D8458307B7943A08
+Int64	180201070C030804007519C20CD964E70BCBD7C204FFBAB40D
+UInt64	180201070C03080400CBD7C204FFBAB40D7519C20CD964E70B
+floats
+Float64	240201070C03080700CBD7C20422822A14EB31BD18289EDB04B1A3460D2BF2FB068467FB11
+Float32	2C0301070C050800010800000022822A14289EDB04CBD7C204ECC8DD04AFD68204B1A3460DD894520CEB31BD18
+BFloat16	180201070C0308040022822A148C904A04731C001BB859430B
+strings
+String	200201070C030806002F823B04218D870411EF66057D7DF6064F461A042E01AA06
+FixedString(5)	140201070C03080300D8B13F07D82DDD0773B92C06
+dates and times
+Date	180201070C03080400CBD7C204A82F26040066F004710BB407
+Date32	1C0201070C03080500015F2707D964E70BCBD7C2042BF2FB06272CF70A
+DateTime	180201070C03080400CBD7C2043F53B008D8458307B7943A08
+DateTime64(0)	1C0201070C030805007BA2BA05D964E70BCBD7C2042BF2FB068425C90A
+DateTime64(3)	1C0201070C0308050040A80204D964E70BCBD7C2042BF2FB06FB95071A
+DateTime64(6)	1C0201070C03080500B89F8606D964E70BCBD7C2042BF2FB06F79DA104
+network and identifiers
+IPv4	1C0201070C03080500CBD7C2043F53B008D8458307F5682308B7943A08
+IPv6	180201070C0308040050C94D05854BD10A18216E096C95BA0E
+UUID	140201070C0308030050C94D05C08DEA136C95BA0E
+Enum8	140201070C030803001FABB112CBD7C2042F537613
+Enum16	140201070C0308030026276604CBD7C204A82F2604
+sketch modes and parameters
+empty, lg_k 12, HLL_4	080201070C030C0000
+empty, lg_k 4, HLL_4	0802010704030C0000
+empty, lg_k 21, HLL_8	0802010715030C0008
+empty, lg_k 12, HLL_6	080201070C030C0004
+list, 8 values	2C0301070C050800010800000081BC5D06862FF90DCBD7C204C1E9170575816607FC2D420A7B65E6082BF2FB06
+set, 9 values	300301070C050800010900000081BC5D06862FF90DCBD7C204C1E91705D216730775816607FC2D420A7B65E6082BF2FB06
+hll, lg_k 4, HLL_4, 100 values	300A0107040008020215EB1DC787F15440000000000000FB3F000000000000000003000000000000000251214121031025
+hll, lg_k 4, HLL_6, 100 values	350A0107040008000615EB1DC787F15440000000000000FB3F0000000000000000000000000000000084301C033118035108C2701000
+hll, lg_k 4, HLL_8, 100 values	380A0107040008000A15EB1DC787F15440000000000000FB3F0000000000000000000000000000000004020307030403060304050202030704
+merging empty states gives an empty sketch
+080201070C030C0000
+integers are widened by value
+Int8 -1 = Int64 -1	1
+Int32 min = Int64 min	1
+UInt8 200 = UInt64 200	1
+UInt8 200 = Int16 200	1
+UInt32 max != Int32 -1	1
+UInt64 max = Int64 -1	1
+floating point values are canonicalized
+-0 = 0	1
+Float32 1.5 = Float64 1.5	1
+Float32 0.1 widened	1
+Float32 0.1 != Float64 0.1	1
+inf != -inf	1
+NaN payloads	1
+types share the hash of their underlying value
+Enum8 max = Int8 127	1
+Date32 -1 = Int32 -1	1
+Date 65535 = UInt16 65535	1
+DateTime 2^31 = UInt32 2^31	1
+IPv4 max = UInt32 max	1
+DateTime64(3) -1 = Int64 -1	1
+String abc = FixedString(3) abc	1
+String abc != FixedString(5) abc	1
+LowCardinality(String) = String	1
diff --git a/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.sql b/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.sql
new file mode 100644
index 000000000000..589ec1b16a7d
--- /dev/null
+++ b/tests/queries/0_stateless/05137_uniq_apache_hll_hash_borders.sql
@@ -0,0 +1,84 @@
+-- Tags: no-fasttest
+-- ^ DataSketches is not built in fast-test builds.
+
+-- Pins how every supported argument type is hashed, on the border values of its range.
+-- Each expected state was produced by the Apache DataSketches C++ library: integers are
+-- widened by value to 64 bits (as Java `update(long)` and Python `update(int)` do),
+-- floats are widened to `double` with `-0.0` and `NaN` canonicalized, and strings and
+-- fixed-size types are hashed as raw bytes. A state fed with up to eight values stays
+-- a coupon list, so the bytes depend only on the values and their order.
+
+SELECT 'integers';
+SELECT 'Int8', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['-128', '-1', '0', '1', '127'], 'Array(Int8)')) AS x) SETTINGS max_threads = 1;
+SELECT 'UInt8', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['0', '127', '128', '255'], 'Array(UInt8)')) AS x) SETTINGS max_threads = 1;
+SELECT 'Int16', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['-32768', '-1', '0', '32767'], 'Array(Int16)')) AS x) SETTINGS max_threads = 1;
+SELECT 'UInt16', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['0', '32767', '32768', '65535'], 'Array(UInt16)')) AS x) SETTINGS max_threads = 1;
+SELECT 'Int32', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['-2147483648', '-1', '0', '2147483647'], 'Array(Int32)')) AS x) SETTINGS max_threads = 1;
+SELECT 'UInt32', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['0', '2147483647', '2147483648', '4294967295'], 'Array(UInt32)')) AS x) SETTINGS max_threads = 1;
+SELECT 'Int64', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['-9223372036854775808', '-1', '0', '9223372036854775807'], 'Array(Int64)')) AS x) SETTINGS max_threads = 1;
+SELECT 'UInt64', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['0', '9223372036854775807', '9223372036854775808', '18446744073709551615'], 'Array(UInt64)')) AS x) SETTINGS max_threads = 1;
+
+SELECT 'floats';
+SELECT 'Float64', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([reinterpretAsFloat64(unhex('0000000000000000')), reinterpretAsFloat64(unhex('0000000000000080')), reinterpretAsFloat64(unhex('000000000000F83F')), reinterpretAsFloat64(unhex('000000000000F87F')), reinterpretAsFloat64(unhex('000000000000F07F')), reinterpretAsFloat64(unhex('000000000000F0FF')), reinterpretAsFloat64(unhex('0100000000000000')), reinterpretAsFloat64(unhex('FFFFFFFFFFFFEF7F'))]) AS x) SETTINGS max_threads = 1;
+SELECT 'Float32', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([reinterpretAsFloat32(unhex('00000000')), reinterpretAsFloat32(unhex('00000080')), reinterpretAsFloat32(unhex('CDCCCC3D')), reinterpretAsFloat32(unhex('0000C03F')), reinterpretAsFloat32(unhex('0000C07F')), reinterpretAsFloat32(unhex('0000807F')), reinterpretAsFloat32(unhex('000080FF')), reinterpretAsFloat32(unhex('FFFF7F7F')), reinterpretAsFloat32(unhex('01000000'))]) AS x) SETTINGS max_threads = 1;
+SELECT 'BFloat16', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([toBFloat16(1.5), toBFloat16(-2.0), toBFloat16(0.5), toBFloat16(256.0)]) AS x) SETTINGS max_threads = 1;
+
+SELECT 'strings';
+SELECT 'String', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['a', 'abc', 'привет', 'a\0b', ' ', '\xFF']) AS x) SETTINGS max_threads = 1;
+SELECT 'FixedString(5)', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['a', 'abc', 'abcde'], 'Array(FixedString(5))')) AS x) SETTINGS max_threads = 1;
+
+SELECT 'dates and times';
+SELECT 'Date', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> toDate(v), [0, 32767, 32768, 65535])) AS x) SETTINGS max_threads = 1;
+SELECT 'Date32', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> toDate32(v), [-25567, -1, 0, 1, 120529])) AS x) SETTINGS max_threads = 1;
+SELECT 'DateTime', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> toDateTime(v, 'UTC'), [0, 2147483647, 2147483648, 4294967295])) AS x) SETTINGS max_threads = 1;
+SELECT 'DateTime64(0)', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> toDateTime64(v, 0, 'UTC'), CAST(['-2208988799', '-1', '0', '1', '1577836800'], 'Array(Int64)'))) AS x) SETTINGS max_threads = 1;
+SELECT 'DateTime64(3)', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> fromUnixTimestamp64Milli(v, 'UTC'), CAST(['-2208988799999', '-1', '0', '1', '1577836800123'], 'Array(Int64)'))) AS x) SETTINGS max_threads = 1;
+SELECT 'DateTime64(6)', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(arrayMap(v -> fromUnixTimestamp64Micro(v, 'UTC'), CAST(['-2208988799999999', '-1', '0', '1', '1577836800123456'], 'Array(Int64)'))) AS x) SETTINGS max_threads = 1;
+
+SELECT 'network and identifiers';
+SELECT 'IPv4', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['0.0.0.0', '127.255.255.255', '128.0.0.0', '192.168.1.1', '255.255.255.255'], 'Array(IPv4)')) AS x) SETTINGS max_threads = 1;
+SELECT 'IPv6', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['::', '::1', '2001:db8::1', 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'], 'Array(IPv6)')) AS x) SETTINGS max_threads = 1;
+SELECT 'UUID', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['00000000-0000-0000-0000-000000000000', '01234567-89ab-cdef-0123-456789abcdef', 'ffffffff-ffff-ffff-ffff-ffffffffffff'], 'Array(UUID)')) AS x) SETTINGS max_threads = 1;
+SELECT 'Enum8', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['min', 'zero', 'max'], 'Array(Enum8(\'min\' = -128, \'zero\' = 0, \'max\' = 127))')) AS x) SETTINGS max_threads = 1;
+SELECT 'Enum16', hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['min', 'zero', 'max'], 'Array(Enum16(\'min\' = -32768, \'zero\' = 0, \'max\' = 32767))')) AS x) SETTINGS max_threads = 1;
+
+SELECT 'sketch modes and parameters';
+SELECT 'empty, lg_k 12, HLL_4', hex(toString(uniqApacheHLLState(number))) FROM numbers(0) SETTINGS max_threads = 1;
+SELECT 'empty, lg_k 4, HLL_4', hex(toString(uniqApacheHLLState(4, 'HLL_4')(number))) FROM numbers(0) SETTINGS max_threads = 1;
+SELECT 'empty, lg_k 21, HLL_8', hex(toString(uniqApacheHLLState(21, 'HLL_8')(number))) FROM numbers(0) SETTINGS max_threads = 1;
+SELECT 'empty, lg_k 12, HLL_6', hex(toString(uniqApacheHLLState(12, 'HLL_6')(number))) FROM numbers(0) SETTINGS max_threads = 1;
+SELECT 'list, 8 values', hex(toString(uniqApacheHLLState(number))) FROM numbers(8) SETTINGS max_threads = 1;
+SELECT 'set, 9 values', hex(toString(uniqApacheHLLState(number))) FROM numbers(9) SETTINGS max_threads = 1;
+SELECT 'hll, lg_k 4, HLL_4, 100 values', hex(toString(uniqApacheHLLState(4, 'HLL_4')(number))) FROM numbers(100) SETTINGS max_threads = 1;
+SELECT 'hll, lg_k 4, HLL_6, 100 values', hex(toString(uniqApacheHLLState(4, 'HLL_6')(number))) FROM numbers(100) SETTINGS max_threads = 1;
+SELECT 'hll, lg_k 4, HLL_8, 100 values', hex(toString(uniqApacheHLLState(4, 'HLL_8')(number))) FROM numbers(100) SETTINGS max_threads = 1;
+
+SELECT 'merging empty states gives an empty sketch';
+SELECT hex(toString(uniqApacheHLLMergeState(s))) FROM (SELECT uniqApacheHLLState(number) AS s FROM numbers(0));
+
+SELECT 'integers are widened by value';
+SELECT 'Int8 -1 = Int64 -1', hex(toString(uniqApacheHLLState(toInt8(-1)))) = hex(toString(uniqApacheHLLState(toInt64(-1))));
+SELECT 'Int32 min = Int64 min', hex(toString(uniqApacheHLLState(toInt32(-2147483648)))) = hex(toString(uniqApacheHLLState(toInt64(-2147483648))));
+SELECT 'UInt8 200 = UInt64 200', hex(toString(uniqApacheHLLState(toUInt8(200)))) = hex(toString(uniqApacheHLLState(toUInt64(200))));
+SELECT 'UInt8 200 = Int16 200', hex(toString(uniqApacheHLLState(toUInt8(200)))) = hex(toString(uniqApacheHLLState(toInt16(200))));
+SELECT 'UInt32 max != Int32 -1', hex(toString(uniqApacheHLLState(toUInt32(4294967295)))) != hex(toString(uniqApacheHLLState(toInt32(-1))));
+SELECT 'UInt64 max = Int64 -1', hex(toString(uniqApacheHLLState(toUInt64(18446744073709551615)))) = hex(toString(uniqApacheHLLState(toInt64(-1))));
+
+SELECT 'floating point values are canonicalized';
+SELECT '-0 = 0', hex(toString(uniqApacheHLLState(toFloat64('-0')))) = hex(toString(uniqApacheHLLState(toFloat64('0'))));
+SELECT 'Float32 1.5 = Float64 1.5', hex(toString(uniqApacheHLLState(toFloat32(1.5)))) = hex(toString(uniqApacheHLLState(toFloat64(1.5))));
+SELECT 'Float32 0.1 widened', hex(toString(uniqApacheHLLState(toFloat32(0.1)))) = hex(toString(uniqApacheHLLState(toFloat64(toFloat32(0.1)))));
+SELECT 'Float32 0.1 != Float64 0.1', hex(toString(uniqApacheHLLState(toFloat32(0.1)))) != hex(toString(uniqApacheHLLState(toFloat64(0.1))));
+SELECT 'inf != -inf', hex(toString(uniqApacheHLLState(toFloat64('inf')))) != hex(toString(uniqApacheHLLState(toFloat64('-inf'))));
+SELECT 'NaN payloads', uniqApacheHLL(x) FROM (SELECT arrayJoin([reinterpretAsFloat64(unhex('000000000000F87F')), reinterpretAsFloat64(unhex('010000000000F87F')), reinterpretAsFloat64(unhex('000000000000F8FF'))]) AS x);
+
+SELECT 'types share the hash of their underlying value';
+SELECT 'Enum8 max = Int8 127', hex(toString(uniqApacheHLLState(CAST('max', 'Enum8(\'max\' = 127)')))) = hex(toString(uniqApacheHLLState(toInt8(127))));
+SELECT 'Date32 -1 = Int32 -1', hex(toString(uniqApacheHLLState(toDate32(toInt32(-1))))) = hex(toString(uniqApacheHLLState(toInt32(-1))));
+SELECT 'Date 65535 = UInt16 65535', hex(toString(uniqApacheHLLState(toDate(65535)))) = hex(toString(uniqApacheHLLState(toUInt16(65535))));
+SELECT 'DateTime 2^31 = UInt32 2^31', hex(toString(uniqApacheHLLState(toDateTime(toUInt32(2147483648), 'UTC')))) = hex(toString(uniqApacheHLLState(toUInt32(2147483648))));
+SELECT 'IPv4 max = UInt32 max', hex(toString(uniqApacheHLLState(toIPv4('255.255.255.255')))) = hex(toString(uniqApacheHLLState(toUInt32(4294967295))));
+SELECT 'DateTime64(3) -1 = Int64 -1', hex(toString(uniqApacheHLLState(fromUnixTimestamp64Milli(toInt64(-1), 'UTC')))) = hex(toString(uniqApacheHLLState(toInt64(-1))));
+SELECT 'String abc = FixedString(3) abc', hex(toString(uniqApacheHLLState('abc'))) = hex(toString(uniqApacheHLLState(toFixedString('abc', 3))));
+SELECT 'String abc != FixedString(5) abc', hex(toString(uniqApacheHLLState('abc'))) != hex(toString(uniqApacheHLLState(toFixedString('abc', 5))));
+SELECT 'LowCardinality(String) = String', hex(toString(uniqApacheHLLState(toLowCardinality('abc')))) = hex(toString(uniqApacheHLLState('abc')));
diff --git a/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.reference b/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.reference
new file mode 100644
index 000000000000..108f9422f1ee
--- /dev/null
+++ b/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.reference
@@ -0,0 +1,20 @@
+empty strings
+0
+080201070C030C0000
+0C0201070C030801002F823B04
+1
+1
+NULL values
+2
+1
+0
+0
+a Nullable argument gives the same state as a plain one
+1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06
+1
+080201070C030C0000
+080201070C030C0000
+080201070C030C0000
+1
+5
+3
diff --git a/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.sql b/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.sql
new file mode 100644
index 000000000000..94282a2c346b
--- /dev/null
+++ b/tests/queries/0_stateless/05138_uniq_apache_hll_null_and_empty.sql
@@ -0,0 +1,50 @@
+-- Tags: no-fasttest
+-- ^ DataSketches is not built in fast-test builds.
+
+-- `NULL` values and empty strings are ignored by the Apache DataSketches implementations in Java, Python
+-- and C++, and by Spark's `hll_sketch_agg`, so a sketch built here must ignore them too.
+-- Unlike `uniq`, an empty string is therefore not counted as a value.
+
+SELECT 'empty strings';
+SELECT uniqApacheHLL(x) FROM (SELECT arrayJoin(['', '', '']) AS x);
+-- Nothing but empty strings leaves the empty sketch.
+SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['', '']) AS x);
+-- The state of 'a' written by the DataSketches C++ library.
+SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['', 'a', '']) AS x) SETTINGS max_threads = 1;
+SELECT
+    (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['a', '', 'b']) AS x))
+  = (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['a', 'b']) AS x))
+SETTINGS max_threads = 1;
+-- A fixed-size string of zero bytes is not empty.
+SELECT uniqApacheHLL(toFixedString('', 3));
+
+SELECT 'NULL values';
+SELECT uniqApacheHLL(x) FROM (SELECT arrayJoin([NULL, 1, NULL, 2, 2]) AS x);
+SELECT uniqApacheHLL(x) FROM (SELECT arrayJoin(['', NULL, 'a']) AS x);
+SELECT uniqApacheHLL(CAST(NULL, 'Nullable(UInt64)')) FROM numbers(3);
+SELECT uniqApacheHLL(toNullable(number)) FROM numbers(0);
+
+SELECT 'a Nullable argument gives the same state as a plain one';
+-- No flag byte of the `Null` combinator: the state is the bare DataSketches sketch, also when every row was `NULL`.
+SELECT hex(toString(uniqApacheHLLState(toNullable(number)))) FROM numbers(5) SETTINGS max_threads = 1;
+SELECT
+    (SELECT hex(toString(uniqApacheHLLState(toNullable(number)))) FROM numbers(5))
+  = (SELECT hex(toString(uniqApacheHLLState(number))) FROM numbers(5))
+SETTINGS max_threads = 1;
+SELECT hex(toString(uniqApacheHLLState(toNullable(number)))) FROM numbers(0);
+SELECT hex(toString(uniqApacheHLLState(CAST(NULL, 'Nullable(UInt64)')))) FROM numbers(3);
+SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['', NULL]) AS x);
+-- NULL rows between values do not change the state.
+SELECT
+    (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([NULL, 1, NULL, 2]) AS x))
+  = (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([toNullable(1), 2]) AS x))
+SETTINGS max_threads = 1;
+-- An external sketch can be imported into a Nullable state type as it is.
+SELECT finalizeAggregation(CAST(unhex('1C0201070C03080500CBD7C2042BF2FB06862FF90D7581660781BC5D06'), 'AggregateFunction(uniqApacheHLL, Nullable(UInt64))'));
+-- Merging Nullable states, including one built from `NULL` values only.
+SELECT uniqApacheHLLMerge(s) FROM
+(
+    SELECT uniqApacheHLLState(toNullable(number)) AS s FROM numbers(3)
+    UNION ALL
+    SELECT uniqApacheHLLState(CAST(NULL, 'Nullable(UInt64)')) AS s FROM numbers(2)
+);
diff --git a/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.reference b/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.reference
new file mode 100644
index 000000000000..52504bd4b578
--- /dev/null
+++ b/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.reference
@@ -0,0 +1,26 @@
+the state type does not depend on the nullability of the argument
+AggregateFunction(uniqApacheHLL, UInt64)
+AggregateFunction(uniqApacheHLL, UInt64)
+AggregateFunction(uniqApacheHLL(14, \'HLL_8\'), String)
+LowCardinality is stripped from the arguments of aggregate functions
+AggregateFunction(uniqApacheHLL, String)
+2
+1
+states of Nullable and plain arguments mix
+6
+the If combinator over Nullable arguments keeps the same state and type
+AggregateFunction(uniqApacheHLL, UInt64)
+AggregateFunction(uniqApacheHLL, UInt64)
+AggregateFunction(uniqApacheHLL, UInt64)
+1
+1
+5
+5
+5
+a declared Nullable state type is interchangeable with the plain one
+AggregateFunction(uniqApacheHLL, Nullable(UInt64))
+4
+4
+tables
+6
+6
diff --git a/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.sql b/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.sql
new file mode 100644
index 000000000000..c75d1cf0153f
--- /dev/null
+++ b/tests/queries/0_stateless/05139_uniq_apache_hll_nullable_state_type.sql
@@ -0,0 +1,63 @@
+-- Tags: no-fasttest
+-- ^ DataSketches is not built in fast-test builds.
+
+-- The state of a `Nullable` argument is the same bare DataSketches sketch as the state of a plain one,
+-- so it also has the same type, and the two can be mixed without a `CAST`.
+
+SELECT 'the state type does not depend on the nullability of the argument';
+SELECT toTypeName(uniqApacheHLLState(toNullable(number))) FROM numbers(1);
+SELECT toTypeName(uniqApacheHLLState(number)) FROM numbers(1);
+SELECT toTypeName(uniqApacheHLLState(14, 'HLL_8')(toNullable(toString(number)))) FROM numbers(1);
+
+SELECT 'LowCardinality is stripped from the arguments of aggregate functions';
+SELECT toTypeName(uniqApacheHLLState(toLowCardinality(toNullable(toString(number))))) FROM numbers(1);
+SELECT uniqApacheHLL(x) FROM (SELECT arrayJoin(CAST(['a', NULL, 'b', 'a', ''], 'Array(LowCardinality(Nullable(String)))')) AS x);
+SELECT
+    (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(CAST(['a', NULL, 'b'], 'Array(LowCardinality(Nullable(String)))')) AS x))
+  = (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin(['a', 'b']) AS x))
+SETTINGS max_threads = 1;
+
+SELECT 'states of Nullable and plain arguments mix';
+SELECT uniqApacheHLLMerge(s) FROM
+(
+    SELECT uniqApacheHLLState(toNullable(number)) AS s FROM numbers(3)
+    UNION ALL
+    SELECT uniqApacheHLLState(number + 3) AS s FROM numbers(3)
+);
+
+SELECT 'the If combinator over Nullable arguments keeps the same state and type';
+SELECT toTypeName(uniqApacheHLLStateIf(toNullable(number), number % 2 = 0)) FROM numbers(1);
+SELECT toTypeName(uniqApacheHLLStateIf(number, number % 2 = 0)) FROM numbers(1);
+SELECT toTypeName(uniqApacheHLLStateIf(number, toNullable(number % 2 = 0))) FROM numbers(1);
+-- The state is the bare sketch of the values 0, 2 and 4, without a flag byte.
+SELECT
+    (SELECT hex(toString(uniqApacheHLLStateIf(toNullable(number), number % 2 = 0))) FROM numbers(5))
+  = (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([0, 2, 4]) AS x))
+SETTINGS max_threads = 1;
+SELECT
+    (SELECT hex(toString(uniqApacheHLLStateIf(number, toNullable(number % 2 = 0)))) FROM numbers(5))
+  = (SELECT hex(toString(uniqApacheHLLState(x))) FROM (SELECT arrayJoin([0, 2, 4]) AS x))
+SETTINGS max_threads = 1;
+SELECT uniqApacheHLLIf(toNullable(number), number % 2 = 0) FROM numbers(10);
+SELECT uniqApacheHLLMergeIf(s, c) FROM (SELECT uniqApacheHLLState(number) AS s, number % 2 = 0 AS c FROM numbers(10) GROUP BY number);
+SELECT uniqApacheHLLMergeIf(s, toNullable(c)) FROM (SELECT uniqApacheHLLState(number) AS s, number % 2 = 0 AS c FROM numbers(10) GROUP BY number);
+
+SELECT 'a declared Nullable state type is interchangeable with the plain one';
+SELECT toTypeName(CAST(uniqApacheHLLState(number), 'AggregateFunction(uniqApacheHLL, Nullable(UInt64))')) FROM numbers(1);
+SELECT finalizeAggregation(CAST(uniqApacheHLLState(number), 'AggregateFunction(uniqApacheHLL, Nullable(UInt64))')) FROM numbers(4);
+SELECT finalizeAggregation(CAST(CAST(uniqApacheHLLState(number), 'AggregateFunction(uniqApacheHLL, Nullable(UInt64))'), 'AggregateFunction(uniqApacheHLL, UInt64)')) FROM numbers(4);
+
+SELECT 'tables';
+DROP TABLE IF EXISTS hll_plain_states;
+CREATE TABLE hll_plain_states (k UInt8, s AggregateFunction(uniqApacheHLL, UInt64)) ENGINE = AggregatingMergeTree ORDER BY k;
+INSERT INTO hll_plain_states SELECT 0 AS k, uniqApacheHLLState(toNullable(number)) FROM numbers(3);
+INSERT INTO hll_plain_states SELECT 0 AS k, uniqApacheHLLState(number + 3) FROM numbers(3);
+SELECT uniqApacheHLLMerge(s) FROM hll_plain_states;
+DROP TABLE hll_plain_states;
+
+DROP TABLE IF EXISTS hll_nullable_states;
+CREATE TABLE hll_nullable_states (k UInt8, s AggregateFunction(uniqApacheHLL, Nullable(UInt64))) ENGINE = AggregatingMergeTree ORDER BY k;
+INSERT INTO hll_nullable_states SELECT 0 AS k, uniqApacheHLLState(toNullable(number)) FROM numbers(3);
+INSERT INTO hll_nullable_states SELECT 0 AS k, uniqApacheHLLState(number + 3) FROM numbers(3);
+SELECT uniqApacheHLLMerge(s) FROM hll_nullable_states;
+DROP TABLE hll_nullable_states;