std::unique_ptr MultiRange::mergeWith()

in velox/type/Filter.cpp [565:645]
67 lines of code
23 McCabe index (conditional complexity)

std::unique_ptr<Filter> MultiRange::mergeWith(const Filter* other) const {
  switch (other->kind()) {
    // Rules of MultiRange with IsNull/IsNotNull
    // 1. MultiRange(nullAllowed=true) AND IS NULL => IS NULL
    // 2. MultiRange(nullAllowed=true) AND IS NOT NULL =>
    // MultiRange(nullAllowed=false)
    // 3. MultiRange(nullAllowed=false) AND IS NULL
    // => ALWAYS FALSE
    // 4. MultiRange(nullAllowed=false) AND IS NOT NULL
    // =>MultiRange(nullAllowed=false)
    case FilterKind::kAlwaysTrue:
    case FilterKind::kAlwaysFalse:
    case FilterKind::kIsNull:
      return other->mergeWith(this);
    case FilterKind::kIsNotNull:
      return this->clone(/*nullAllowed=*/false);
    case FilterKind::kDoubleRange:
    case FilterKind::kFloatRange:
      // TODO: Implement
      VELOX_UNREACHABLE();
    case FilterKind::kBytesValues:
    case FilterKind::kBytesRange:
    case FilterKind::kMultiRange: {
      bool bothNullAllowed = nullAllowed_ && other->testNull();
      bool bothNanAllowed = nanAllowed_;
      std::vector<const Filter*> otherFilters;

      if (other->kind() == FilterKind::kMultiRange) {
        auto multiRangeOther = static_cast<const MultiRange*>(other);
        for (auto const& filterOther : multiRangeOther->filters()) {
          otherFilters.emplace_back(filterOther.get());
        }
        bothNanAllowed = bothNanAllowed && multiRangeOther->nanAllowed();
      } else {
        otherFilters.emplace_back(other);
      }

      std::vector<std::string> byteValues;
      std::vector<std::unique_ptr<Filter>> merged;
      merged.reserve(this->filters().size() + otherFilters.size());

      for (auto const& filter : this->filters()) {
        for (auto const& filterOther : otherFilters) {
          auto innerMerged = filter->mergeWith(filterOther);
          switch (innerMerged->kind()) {
            case FilterKind::kAlwaysFalse:
            case FilterKind::kIsNull:
              continue;
            case FilterKind::kBytesValues: {
              auto mergedBytesValues =
                  static_cast<const BytesValues*>(innerMerged.get());
              byteValues.reserve(
                  byteValues.size() + mergedBytesValues->values().size());
              for (const auto& value : mergedBytesValues->values()) {
                byteValues.emplace_back(value);
              }
            }
            default:
              merged.emplace_back(innerMerged.release());
          }
        }
      }

      if (!byteValues.empty()) {
        merged.emplace_back(std::make_unique<BytesValues>(
            std::move(byteValues), bothNullAllowed));
      }

      if (merged.empty()) {
        return nullOrFalse(bothNullAllowed);
      } else if (merged.size() == 1) {
        return merged.front()->clone(bothNullAllowed);
      } else {
        return std::make_unique<MultiRange>(
            std::move(merged), bothNullAllowed, bothNanAllowed);
      }
    }
    default:
      VELOX_UNREACHABLE();
  }
}