Fix HLL union estimate/serialization regression from lazy KxQ rebuild

hll/include/Hll8Array-internal.hpp

@@ -98,6 +98,10 @@ void Hll8Array<A>::internalCouponUpdate(uint32_t coupon) {
 
   const uint8_t curVal = this->hllByteArr_[slotNo];
   if (newVal > curVal) {
+    // A prior HLL merge may have deferred rebuilding the estimator state. Rebuild only when
+    // this coupon changes a register, immediately before the incremental update needs KxQ and
+    // numAtCurMin_. Duplicate coupons therefore preserve the lazy-merge optimization.
+    if (this->rebuild_kxq_curmin_) this->check_rebuild_kxq_cur_min();
     this->hllByteArr_[slotNo] = newVal;
     this->hipAndKxQIncrementalUpdate(curVal, newVal);
     this->numAtCurMin_ -= curVal == 0; // interpret numAtCurMin as num zeros

hll/include/HllArray-internal.hpp

@@ -64,10 +64,14 @@ HllArray<A>::HllArray(const HllArray& other, target_hll_type tgtHllType) :
 
 template<typename A>
 HllArray<A>* HllArray<A>::copyAs(target_hll_type tgtHllType) const {
-  // we may need to recompute KxQ and curMin data for a union gadget,
-  // so only use a direct copy if we have a valid sketch
-  if (tgtHllType == this->getTgtHllType() && !this->isRebuildKxqCurminFlag()) {
-    return static_cast<HllArray*>(copy());
+  if (tgtHllType == this->getTgtHllType()) {
+    // Preserve lazy merging in the source, but make every same-type result use the direct-sum
+    // rebuild. Replaying registers through the conversion constructor produces slightly
+    // different floating-point KxQ values, making serialization depend on when the lazy state
+    // escaped through copyAs().
+    HllArray* result = static_cast<HllArray*>(copy());
+    result->check_rebuild_kxq_cur_min();
+    return result;
   }
 
   // the factory methods replay the coupons and will always rebuild
@@ -465,6 +469,10 @@ bool HllArray<A>::isCompact() const {
 
 template<typename A>
 bool HllArray<A>::isEmpty() const {
+  // mergeHll() is only called with non-empty sketches and sets this flag after updating the
+  // register array. The cached curMin_/numAtCurMin_ may still have their empty-sketch values,
+  // but a pending rebuild therefore proves that this array is not empty.
+  if (rebuild_kxq_curmin_) return false;
   const uint32_t configK = 1 << this->lgConfigK_;
   return (curMin_ == 0) && (numAtCurMin_ == configK);
 }

hll/test/HllUnionTest.cpp

@@ -314,4 +314,73 @@ TEST_CASE("hll union: check hll to hll", "[hll_union]") {
   union_two_sketches_with_overlap(1000000, 11, HLL_4);
 }
 
+static hll_sketch::vector_bytes serialize_flat_union(
+    const hll_sketch& first, const hll_sketch& second, const hll_sketch& third) {
+  hll_union u(8);
+  u.update(first);
+  u.update(second);
+  u.update(third);
+  return u.get_result(HLL_8).serialize_updatable();
+}
+
+static hll_sketch::vector_bytes serialize_nested_union(
+    const hll_sketch& first, const hll_sketch& second, const hll_sketch& third) {
+  hll_union prefix(8);
+  prefix.update(first);
+  prefix.update(second);
+
+  hll_union u(8);
+  u.update(prefix.get_result(HLL_8));
+  u.update(third);
+  return u.get_result(HLL_8).serialize_updatable();
+}
+
+// Regression test for a lazy KxQ/curMin rebuild bug in HLL union.
+// When an HLL-mode sketch with a larger lgConfigK is merged into a union first, the union
+// downsamples it via mergeHll(), which defers the KxQ/curMin recompute (rebuild_kxq_curmin_).
+// That left numAtCurMin_/curMin_ at their empty-sketch defaults, so is_empty() wrongly reported
+// the populated gadget as empty and the next update discarded the accumulated result; a scalar
+// update afterwards instead corrupted the HIP accumulator (read because the downsampled gadget
+// keeps oooFlag_ == false). Both made the estimate collapse to roughly a single input's
+// cardinality. Estimates and serialized results must be independent of merge order.
+TEST_CASE("hll union: mixed lgConfigK merge order independence", "[hll_union]") {
+  const uint64_t n = 100000; // large enough to put all inputs in HLL mode at these lgConfigK
+  hll_sketch a(15);
+  hll_sketch b(8);
+  for (uint64_t i = 0; i < n; ++i) a.update(i);
+  for (uint64_t i = n; i < 2 * n; ++i) b.update(i);
+  const double truth = 2.0 * n;
+
+  // A single downsampling merge must not report the populated gadget as empty.
+  hll_union u0(8);
+  u0.update(a);
+  REQUIRE_FALSE(u0.is_empty());
+
+  // Larger-lgConfigK sketch merged first (forces downsample-first), then the smaller one.
+  hll_union u1(8);
+  u1.update(a);
+  u1.update(b);
+  REQUIRE(u1.get_estimate() == Approx(truth).epsilon(0.1));
+
+  // Smaller-lgConfigK first must give the same answer.
+  hll_union u2(8);
+  u2.update(b);
+  u2.update(a);
+  REQUIRE(u2.get_estimate() == Approx(truth).epsilon(0.1));
+
+  // Scalar updates after a downsample-first merge must also accumulate correctly.
+  hll_union u3(8);
+  u3.update(a);
+  for (uint64_t i = n; i < 2 * n; ++i) u3.update(i);
+  REQUIRE(u3.get_estimate() == Approx(truth).epsilon(0.1));
+
+  // Extracting an intermediate lazy result must use the same KxQ rebuild as the flat union.
+  // Otherwise equivalent register arrays can serialize different floating-point KxQ values.
+  hll_sketch c(11);
+  for (uint64_t i = n / 2; i < n + n / 2; ++i) c.update(i);
+  REQUIRE(serialize_nested_union(b, c, a) == serialize_flat_union(b, c, a));
+  REQUIRE(serialize_nested_union(a, c, b) == serialize_flat_union(a, c, b));
+  REQUIRE(serialize_nested_union(a, b, c) == serialize_flat_union(a, b, c));
+}
+
 } /* namespace datasketches */