UniqApacheHll fixes based on feedback

src/AggregateFunctions/AggregateFunctionUniqApacheHLL.cpp

@@ -21,10 +21,11 @@ static AggregateFunctionPtr createAggregateFunctionUniqApacheHLL(
 {
     uint8_t lg_config_k = 12;
     datasketches::target_hll_type target_type = datasketches::HLL_4;
+    bool use_composite_estimate = false;
 
-    if (params.size() > 2)
+    if (params.size() > 3)
         throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
-            "Aggregate function {} accepts at most two parameters (lg_k, type).", name);
+            "Aggregate function {} accepts at most three parameters (lg_k, type, estimator).", name);
 
     if (!params.empty())
     {
@@ -35,7 +36,7 @@ static AggregateFunctionPtr createAggregateFunctionUniqApacheHLL(
         lg_config_k = static_cast<uint8_t>(lg_k_param);
     }
 
-    if (params.size() == 2)
+    if (params.size() >= 2)
     {
         if (params[1].getType() != Field::Types::String)
             throw Exception(ErrorCodes::BAD_ARGUMENTS,
@@ -53,42 +54,58 @@ static AggregateFunctionPtr createAggregateFunctionUniqApacheHLL(
                 "Parameter type for aggregate function {} must be one of 'HLL_4', 'HLL_6', 'HLL_8'.", name);
     }
 
+    if (params.size() == 3)
+    {
+        if (params[2].getType() != Field::Types::String)
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Parameter estimator for aggregate function {} must be a string.", name);
+
+        const String estimator_param = params[2].safeGet<String>();
+        if (estimator_param == "DEFAULT")
+            use_composite_estimate = false;
+        else if (estimator_param == "COMPOSITE")
+            use_composite_estimate = true;
+        else
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Parameter estimator for aggregate function {} must be one of 'DEFAULT', 'COMPOSITE'.", 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);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDateTime64::FieldType>>(lg_config_k, target_type, use_composite_estimate, 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));
+            argument_type, lg_config_k, target_type, use_composite_estimate, 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);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDate::FieldType>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isDate32())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDate32::FieldType>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDate32::FieldType>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isDateTime())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDateTime::FieldType>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeDateTime::FieldType>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isStringOrFixedString())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<String>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<String>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isUUID())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeUUID::FieldType>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeUUID::FieldType>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isIPv4())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv4::FieldType>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv4::FieldType>>(lg_config_k, target_type, use_composite_estimate, argument_types, params);
     if (which.isIPv6())
-        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv6::FieldType>>(lg_config_k, target_type, argument_types, params);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<DataTypeIPv6::FieldType>>(lg_config_k, target_type, use_composite_estimate, 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);
+        return std::make_shared<AggregateFunctionUniqApacheHLL<UInt8>>(lg_config_k, target_type, use_composite_estimate, 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.",
@@ -105,21 +122,27 @@ The `-State` and `-Merge` combinators exchange Apache DataSketches HLL sketches
 
 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.
+An unsigned `UInt8`, `UInt16` or `UInt32` value is hashed as the same number as a 64-bit integer, as Java and Spark hash a number that they hold as a `long`.
+The C++ library hashes `uint8_t`, `uint16_t` and `uint32_t` through the signed type of the same width, so for the values from 128, 32768 and 2147483648 respectively (and for `Date` from 32768, `DateTime` from 2147483648, and `IPv4` from 128.0.0.0) its sketch is another one.
+To get the sketch of the C++ library, cast the argument to the signed type of the same width with a cast that wraps: `toInt8(x)`, `toInt16(x)` or `toInt32(x)`; for `Date` use `toInt16(toUInt16(x))`, for `DateTime` use `toInt32(x)`, and for `IPv4` use `toInt32(toUInt32(x))`.
+Do not use `accurateCast`, which rejects these values.
 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 can switch from the HIP estimator to the less accurate composite estimator, and the HIP estimate of a union of small sketches depends on the order of the merges, so with the default estimator results can depend on partitioning across threads, parts and shards.
+The `estimator` parameter `'COMPOSITE'` makes the result depend only on the registers of the sketch and not on the order of the inserts and of the merges, at the cost of a lower accuracy for sketches of a few hundred distinct values. It does not change the state.
 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::Syntax syntax = "uniqApacheHLL([lg_k, [type, [estimator]]])(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"}},
+        {"estimator", "Optional. The estimator used for the result: 'DEFAULT' (the estimator of the library, which can depend on the order of the inserts and of the merges) or 'COMPOSITE' (depends only on the registers of the sketch). Default: 'DEFAULT'.", {"String"}},
     };
     FunctionDocumentation::ReturnedValue returned_value = {"Returns the approximate number of distinct values.", {"UInt64"}};
     FunctionDocumentation::Examples examples = {
@@ -130,7 +153,9 @@ Merging a lower-resolution sketch permanently lowers the result's resolution, re
     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 };
+    /// The HIP estimate and the bytes of the sketch depend on the order of the updates and of the merges, so the planner must not
+    /// remove an `ORDER BY` below this function.
+    AggregateFunctionProperties properties = { .returns_default_when_only_null = true, .is_order_dependent = true };
 
     factory.registerFunction("uniqApacheHLL", {createAggregateFunctionUniqApacheHLL, documentation, properties});
 }

src/AggregateFunctions/AggregateFunctionUniqApacheHLL.h

@@ -4,14 +4,17 @@
 
 #if USE_DATASKETCHES
 
+#include <AggregateFunctions/AggregateFunctionFactory.h>
 #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 <Common/transformEndianness.h>
 #include <Core/Field.h>
 #include <Core/UUID.h>
+#include <DataTypes/DataTypeAggregateFunction.h>
 #include <DataTypes/DataTypeDate.h>
 #include <DataTypes/DataTypeDate32.h>
 #include <DataTypes/DataTypeDateTime.h>
@@ -72,7 +75,7 @@ class HllSketchData
         if (sk_union && sk_update)
         {
             sk_union->update(*sk_update);
-            sk_update.reset();
+            sk_update = nullptr;
         }
     }
 
@@ -88,38 +91,47 @@ class HllSketchData
         getSkUpdate(lg_config_k, tgt_type)->update(static_cast<const void *>(data), size);
     }
 
-    UInt64 size() const
+    /// With `use_composite_estimate` the estimate is computed from the registers of the sketch only, so it does not depend on the order
+    /// of the inserts and of the merges. The estimate of the library is the HIP estimate while the sketch is in order, which is more
+    /// accurate for a small sketch, but it depends on that order.
+    UInt64 size(bool use_composite_estimate) const
     {
         foldUpdateIntoUnionIfNeeded();
 
         /// Rounding preserves exact cardinalities despite floating-point error.
         if (sk_union)
-            return static_cast<UInt64>(std::llround(sk_union->get_estimate()));
+            return static_cast<UInt64>(std::llround(use_composite_estimate ? sk_union->get_composite_estimate() : sk_union->get_estimate()));
         if (sk_update)
-            return static_cast<UInt64>(std::llround(sk_update->get_estimate()));
+            return static_cast<UInt64>(std::llround(use_composite_estimate ? sk_update->get_composite_estimate() : sk_update->get_estimate()));
         return 0;
     }
 
     void merge(const HllSketchData & rhs, uint8_t lg_config_k)
     {
-        if (!rhs.sk_update && !rhs.sk_union)
+        const bool rhs_has_update = rhs.sk_update && !rhs.sk_update->is_empty();
+        if (!rhs_has_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 rows that are already pending in this state come before `rhs`. They can be there after a merge, for example when
+        /// the rows of the tail of a block are added to a state that has already merged the lookup tables of the block.
+        foldUpdateIntoUnionIfNeeded();
 
+        /// `rhs` may hold both sketches; take both without modifying it, in the order of `foldUpdateIntoUnionIfNeeded`: the union, then the
+        /// updates. The registers of the result do not depend on the order, but the HIP estimate and the bytes of the sketch do.
         /// 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();
+        if (rhs_has_update)
+            u->update(*rhs.sk_update);
     }
 
-    /// You can only call this for an empty object.
-    void read(ReadBuffer & in, uint8_t lg_config_k)
+    /// You can only call this for an empty object. The sketch is kept as it was written, with its own resolution and
+    /// storage type: reading a state must not lose what writing it back would then persist. It is folded into a union,
+    /// at the declared resolution, only when it is merged.
+    void read(ReadBuffer & in)
     {
         datasketches::hll_sketch::vector_bytes bytes;
         readVectorBinary(bytes, in);
@@ -129,7 +141,7 @@ class HllSketchData
         try
         {
             auto sk = datasketches::hll_sketch::deserialize(bytes.data(), bytes.size());
-            getSkUnion(lg_config_k)->update(std::move(sk));
+            sk_update = std::make_unique<datasketches::hll_sketch>(std::move(sk));
         }
         catch (const DB::Exception &)
         {
@@ -169,16 +181,19 @@ class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper
 
     uint8_t lg_config_k;
     datasketches::target_hll_type target_type;
+    bool use_composite_estimate;
 
 public:
     AggregateFunctionUniqApacheHLL(
         uint8_t lg_config_k_,
         datasketches::target_hll_type target_type_,
+        bool use_composite_estimate_,
         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_)
+        , use_composite_estimate(use_composite_estimate_)
     {
     }
 
@@ -199,7 +214,7 @@ class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper
 
             /// Other DataSketches implementations ignore empty strings, and the sketch of a mixed
             /// input must be the same as theirs.
-            if (value.size() == 0)
+            if (value.empty())
                 return;
 
             data.insertData(value.data(), value.size(), lg_config_k, target_type);
@@ -210,11 +225,11 @@ class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper
 
             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)),
-                };
+                /// The two halves of a `UUID` are numbers in host order; the canonical bytes are their big-endian
+                /// representation on every platform.
+                UInt64 halves[2] = {UUIDHelpers::getHighBytes(value), UUIDHelpers::getLowBytes(value)};
+                transformEndianness<std::endian::big>(halves[0]);
+                transformEndianness<std::endian::big>(halves[1]);
                 data.insertData(reinterpret_cast<const char *>(halves), sizeof(halves), lg_config_k, target_type);
             }
             else if constexpr (std::is_same_v<T, IPv6>)
@@ -249,6 +264,21 @@ class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper
                 [](const auto & lhs, const auto & rhs_type) { return removeNullable(lhs)->equals(*removeNullable(rhs_type)); });
     }
 
+    /// The parameters do not change what a state means, so every spelling of them is the same type wherever a common type is
+    /// needed (`UNION ALL`, arrays, `if`) and wherever a state is accepted by the name of its type. The default parameters
+    /// stand for all of them.
+    DataTypePtr getNormalizedStateType() const override
+    {
+        DataTypes normalized_argument_types;
+        normalized_argument_types.reserve(this->argument_types.size());
+        for (const auto & type : this->argument_types)
+            normalized_argument_types.emplace_back(type->getNormalizedType());
+
+        AggregateFunctionProperties properties;
+        auto function = AggregateFunctionFactory::instance().get("uniqApacheHLL", NullsAction::EMPTY, normalized_argument_types, {}, properties);
+        return std::make_shared<DataTypeAggregateFunction>(function, normalized_argument_types, Array{});
+    }
+
     void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override
     {
         this->data(place).merge(this->data(rhs), lg_config_k);
@@ -261,12 +291,12 @@ class AggregateFunctionUniqApacheHLL final : public IAggregateFunctionDataHelper
 
     void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
     {
-        this->data(place).read(buf, lg_config_k);
+        this->data(place).read(buf);
     }
 
     void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
     {
-        assert_cast<ColumnUInt64 &>(to).getData().push_back(this->data(place).size());
+        assert_cast<ColumnUInt64 &>(to).getData().push_back(this->data(place).size(use_composite_estimate));
     }
 };