feat: Add SetDigest result verifier (#15619)

Summary:
Pull Request resolved: #15619

feat: Add SetDigest result verifier to be used in functions with SetDigest output

Differential Revision: D87835419

Author: natashasehgal

velox/functions/prestosql/fuzzer/CMakeLists.txt

@@ -19,6 +19,7 @@ target_link_libraries(
   velox_core
   velox_aggregation_fuzzer_base
   velox_exec_test_lib
+  velox_functions_lib
   velox_vector
   velox_vector_test_lib
 )

velox/functions/prestosql/fuzzer/SetDigestResultVerifier.h

@@ -0,0 +1,188 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include "velox/common/memory/HashStringAllocator.h"
+#include "velox/core/PlanNode.h"
+#include "velox/exec/fuzzer/ResultVerifier.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/functions/lib/SetDigest.h"
+#include "velox/vector/ComplexVector.h"
+
+namespace facebook::velox::exec::test {
+
+class SetDigestResultVerifier : public ResultVerifier {
+ public:
+  bool supportsCompare() override {
+    return true;
+  }
+
+  bool supportsVerify() override {
+    return false;
+  }
+
+  void initialize(
+      const std::vector<RowVectorPtr>& /*input*/,
+      const std::vector<core::ExprPtr>& /*projections*/,
+      const std::vector<std::string>& groupingKeys,
+      const core::AggregationNode::Aggregate& /*aggregate*/,
+      const std::string& aggregateName) override {
+    keys_ = groupingKeys;
+    resultName_ = aggregateName;
+  }
+
+  void initializeWindow(
+      const std::vector<RowVectorPtr>& /*input*/,
+      const std::vector<core::ExprPtr>& /*projections*/,
+      const std::vector<std::string>& /*partitionByKeys*/,
+      const std::vector<SortingKeyAndOrder>& /*sortingKeysAndOrders*/,
+      const core::WindowNode::Function& /*function*/,
+      const std::string& /*frame*/,
+      const std::string& windowName) override {
+    keys_ = {"row_number"};
+    resultName_ = windowName;
+  }
+
+  bool compare(const RowVectorPtr& result, const RowVectorPtr& altResult)
+      override {
+    VELOX_CHECK_EQ(result->size(), altResult->size());
+
+    auto projection = keys_;
+    projection.push_back(resultName_);
+
+    auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
+    auto builder = PlanBuilder(planNodeIdGenerator).values({result});
+    if (!keys_.empty()) {
+      builder = builder.orderBy(keys_, false);
+    }
+    auto sortByKeys = builder.project(projection).planNode();
+    auto sortedResult =
+        AssertQueryBuilder(sortByKeys).copyResults(result->pool());
+
+    builder = PlanBuilder(planNodeIdGenerator).values({altResult});
+    if (!keys_.empty()) {
+      builder = builder.orderBy(keys_, false);
+    }
+    sortByKeys = builder.project(projection).planNode();
+    auto sortedAltResult =
+        AssertQueryBuilder(sortByKeys).copyResults(altResult->pool());
+
+    VELOX_CHECK_EQ(sortedResult->size(), sortedAltResult->size());
+    auto size = sortedResult->size();
+    for (auto i = 0; i < size; i++) {
+      auto resultIsNull = sortedResult->childAt(resultName_)->isNullAt(i);
+      auto altResultIsNull = sortedAltResult->childAt(resultName_)->isNullAt(i);
+      if (resultIsNull || altResultIsNull) {
+        VELOX_CHECK(
+            resultIsNull && altResultIsNull,
+            "Null mismatch at row {}: result={}, altResult={}",
+            i,
+            resultIsNull,
+            altResultIsNull);
+        continue;
+      }
+
+      auto resultValue = sortedResult->childAt(resultName_)
+                             ->as<SimpleVector<StringView>>()
+                             ->valueAt(i);
+      auto altResultValue = sortedAltResult->childAt(resultName_)
+                                ->as<SimpleVector<StringView>>()
+                                ->valueAt(i);
+
+      if (resultValue == altResultValue) {
+        continue;
+      }
+
+      checkEquivalentSetDigest(resultValue, altResultValue);
+    }
+    return true;
+  }
+
+  bool verify(const RowVectorPtr& /*result*/) override {
+    VELOX_UNSUPPORTED();
+  }
+
+  void reset() override {
+    keys_.clear();
+    resultName_.clear();
+  }
+
+ private:
+  void checkEquivalentSetDigest(
+      const StringView& result,
+      const StringView& altResult) {
+    auto pool = memory::memoryManager()->addLeafPool();
+    HashStringAllocator allocator(pool.get());
+
+    facebook::velox::functions::SetDigest resultDigest(&allocator);
+    facebook::velox::functions::SetDigest altResultDigest(&allocator);
+
+    // Deserialize SetDigests
+    try {
+      resultDigest.deserialize(result.data(), result.size());
+      altResultDigest.deserialize(altResult.data(), altResult.size());
+    } catch (const std::exception& e) {
+      VELOX_FAIL("Failed to deserialize SetDigest: {}", e.what());
+    }
+
+    // Extract cardinality and exactness mode
+    auto resultCardinality = resultDigest.cardinality();
+    auto altResultCardinality = altResultDigest.cardinality();
+    bool resultIsExact = resultDigest.isExact();
+    bool altResultIsExact = altResultDigest.isExact();
+
+    // For exact mode, cardinalities must match exactly
+    if (resultIsExact && altResultIsExact) {
+      VELOX_CHECK_EQ(
+          resultCardinality,
+          altResultCardinality,
+          "SetDigest exact cardinality mismatch: {} vs {}",
+          resultCardinality,
+          altResultCardinality);
+      return;
+    }
+
+    // For approximate mode or mixed modes, compare with error tolerance.
+    if (resultCardinality == 0 && altResultCardinality == 0) {
+      return;
+    }
+
+    // Calculate relative error
+    double maxCardinality = std::max(
+        static_cast<double>(resultCardinality),
+        static_cast<double>(altResultCardinality));
+    double errorRate = std::abs(
+                           static_cast<double>(resultCardinality) -
+                           static_cast<double>(altResultCardinality)) /
+        maxCardinality;
+
+    VELOX_CHECK_LT(
+        errorRate,
+        kApproximateErrorTolerance,
+        "SetDigest cardinality mismatch: {} vs {} (error: {:.2%})",
+        resultCardinality,
+        altResultCardinality,
+        errorRate);
+  }
+
+  static constexpr double kApproximateErrorTolerance = 0.05;
+
+  std::vector<std::string> keys_;
+  std::string resultName_;
+};
+
+} // namespace facebook::velox::exec::test