fix: Support merging HUGEINT value filters

velox/connectors/hive/tests/HiveConnectorTest.cpp

@@ -631,10 +631,9 @@ TEST_F(HiveConnectorTest, extractFiltersFromRemainingFilter) {
       extractFiltersFromRemainingFilter(expr, &evaluator, filters, sampleRate);
   ASSERT_EQ(sampleRate, 1);
   ASSERT_GT(filters.count(Subfield("c2")), 0);
-  // Change these once HUGEINT filter merge is fixed.
-  ASSERT_TRUE(remaining);
-  ASSERT_EQ(
-      remaining->toString(), "not(lt(ROW[\"c2\"],cast(0 as DECIMAL(20, 0))))");
+  auto expectedFilter = exec::betweenHugeint(0, 1);
+  ASSERT_TRUE(expectedFilter->testingEquals(*filters.at(Subfield("c2"))));
+  ASSERT_FALSE(remaining);
 
   // parseExpr gives AND/OR with 2 arguments.  We need to construct the node
   // manually to have more than 2.

velox/type/Filter.cpp

@@ -1909,6 +1909,34 @@ std::unique_ptr<Filter> TimestampRange::mergeWith(const Filter* other) const {
   }
 }
 
+std::unique_ptr<Filter> HugeintRange::mergeWith(const Filter* other) const {
+  switch (other->kind()) {
+    case FilterKind::kAlwaysTrue:
+    case FilterKind::kAlwaysFalse:
+    case FilterKind::kIsNull:
+      return other->mergeWith(this);
+    case FilterKind::kIsNotNull:
+      return this->clone(false);
+    case FilterKind::kHugeintRange: {
+      const bool bothNullAllowed = nullAllowed_ && other->testNull();
+      const auto* otherRange = static_cast<const HugeintRange*>(other);
+
+      const auto lower = std::max(lower_, otherRange->lower_);
+      const auto upper = std::min(upper_, otherRange->upper_);
+
+      if (lower <= upper) {
+        return std::make_unique<HugeintRange>(lower, upper, bothNullAllowed);
+      }
+
+      return nullOrFalse(bothNullAllowed);
+    }
+    case FilterKind::kHugeintValuesUsingHashTable:
+      return other->mergeWith(this);
+    default:
+      VELOX_UNREACHABLE();
+  }
+}
+
 std::unique_ptr<Filter> NegatedBigintRange::mergeWith(
     const Filter* other) const {
   switch (other->kind()) {
@@ -2100,6 +2128,32 @@ std::unique_ptr<Filter> BigintValuesUsingHashTable::mergeWith(
   return createBigintValues(valuesToKeep, bothNullAllowed);
 }
 
+std::unique_ptr<Filter> HugeintValuesUsingHashTable::mergeWith(
+    const Filter* other) const {
+  switch (other->kind()) {
+    case FilterKind::kAlwaysTrue:
+    case FilterKind::kAlwaysFalse:
+    case FilterKind::kIsNull:
+      return other->mergeWith(this);
+    case FilterKind::kIsNotNull:
+      return this->clone(false);
+    case FilterKind::kHugeintRange:
+    case FilterKind::kHugeintValuesUsingHashTable: {
+      const bool bothNullAllowed = nullAllowed_ && other->testNull();
+      std::vector<int128_t> valuesToKeep;
+      valuesToKeep.reserve(values_.size());
+      for (const auto value : values_) {
+        if (other->testInt128(value)) {
+          valuesToKeep.push_back(value);
+        }
+      }
+      return createHugeintValues(valuesToKeep, bothNullAllowed);
+    }
+    default:
+      VELOX_UNREACHABLE();
+  }
+}
+
 std::unique_ptr<Filter> BigintValuesUsingBitmask::mergeWith(
     const Filter* other) const {
   switch (other->kind()) {

velox/type/Filter.h

@@ -980,6 +980,8 @@ class HugeintRange final : public Filter {
         nullAllowed_ ? "with nulls" : "no nulls");
   }
 
+  std::unique_ptr<Filter> mergeWith(const Filter* other) const final;
+
   bool testingEquals(const Filter& other) const final;
 
  private:
@@ -1202,6 +1204,8 @@ class HugeintValuesUsingHashTable final : public Filter {
 
   bool testInt128(const int128_t& value) const final;
 
+  std::unique_ptr<Filter> mergeWith(const Filter* other) const final;
+
   bool testingEquals(const Filter& other) const final;
 
   int128_t min() const {

velox/type/tests/FilterTest.cpp

@@ -271,6 +271,65 @@
   EXPECT_FALSE(filter->testInt128(0));
 }
 
+TEST(FilterTest, mergeWithHugeintValuesUsingHashTable) {
+  auto valueAt = [](uint64_t highBits) {
+    return HugeInt::build(highBits, /*lowBits=*/42);
+  };
+
+  const auto value1 = valueAt(1);
+  const auto value2 = valueAt(2);
+  const auto value3 = valueAt(3);
+  const auto value4 = valueAt(4);
+
+  auto test =
+      [](const Filter& left, const Filter& right, const Filter& expected) {
+        auto merged = left.mergeWith(&right);
+        ASSERT_TRUE(merged->testingEquals(expected));
+        auto reverseMerged = right.mergeWith(&left);
+        ASSERT_TRUE(reverseMerged->testingEquals(expected));
+      };
+
+  {
+    SCOPED_TRACE("HugeintValuesUsingHashTable");
+    auto left = createHugeintValues({value1, value2, value3}, true);
+    auto right = createHugeintValues({value2, value4}, false);
+    auto expected = createHugeintValues({value2}, false);
+    test(*left, *right, *expected);
+  }
+
+  {
+    SCOPED_TRACE("HugeintRange");
+    auto values = createHugeintValues({value1, value2, value3}, true);
+    auto range = betweenHugeint(value2, value4, true);
+    auto expected = createHugeintValues({value2, value3}, true);
+    test(*values, *range, *expected);
+  }
+
+  {
+    SCOPED_TRACE("HugeintRange intersection");
+    auto left = betweenHugeint(value1, value3, true);
+    auto right = betweenHugeint(value2, value4, false);
+    auto expected = betweenHugeint(value2, value3, false);
+    test(*left, *right, *expected);
+  }
+
+  {
+    SCOPED_TRACE("Disjoint ranges with nulls");
+    auto left = betweenHugeint(value1, value1, true);
+    auto right = betweenHugeint(value2, value4, true);
+    IsNull expected;
+    test(*left, *right, expected);
+  }
+
+  {
+    SCOPED_TRACE("Disjoint values with nulls");
+    auto left = createHugeintValues({value1}, true);
+    auto right = createHugeintValues({value2}, true);
+    IsNull expected;
+    test(*left, *right, expected);
+  }
+}
+
 TEST(FilterTest, negatedBigintRange) {
   auto filter = notEqual(1, false);
   EXPECT_FALSE(filter->testNull());