private def estimateNumericComparison()

in flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimator.scala [761:905]
114 lines of code
50 McCabe index (conditional complexity)

  private def estimateNumericComparison(
      op: SqlOperator,
      left: RexInputRef,
      right: RexInputRef): Option[Double] = {
    val selectivityWithoutStats = op match {
      case EQUALS => defaultEqualsSelectivity
      case _ => defaultComparisonSelectivity
    }
    val leftInterval = mq.getColumnInterval(this.rel, left.getIndex)
    val rightInterval = mq.getColumnInterval(this.rel, right.getIndex)
    if (
      leftInterval == null ||
      leftInterval == ValueInterval.infinite ||
      rightInterval == null ||
      rightInterval == ValueInterval.infinite
    ) {
      return selectivityWithoutStats
    } else if (leftInterval == ValueInterval.empty || rightInterval == ValueInterval.empty) {
      return Some(0.0)
    }

    val leftNullCount = mq.getColumnNullCount(this.rel, left.getIndex)
    val rightNullCount = mq.getColumnNullCount(this.rel, right.getIndex)
    if (leftNullCount == null || rightNullCount == null) {
      return selectivityWithoutStats
    }

    val bitSetOfLeftInputRef = ImmutableBitSet.of(left.getIndex)
    val leftNdv = mq.getDistinctRowCount(this.rel, bitSetOfLeftInputRef, null)
    val bitSetOfRightInputRef = ImmutableBitSet.of(right.getIndex)
    val rightNdv = mq.getDistinctRowCount(this.rel, bitSetOfRightInputRef, null)
    if (
      op.equals(EQUALS) &&
      (!(leftInterval.isInstanceOf[FiniteValueInterval]
        && rightInterval.isInstanceOf[FiniteValueInterval])) ||
      leftNdv == null || rightNdv == null
    ) {
      return defaultEqualsSelectivity
    }
    // left interval
    val (leftMin, leftIncludeMin) = leftInterval match {
      case hasLower: WithLower => (comparableToDouble(hasLower.lower), hasLower.includeLower)
      case _ => (null, true)
    }
    val (leftMax, leftIncludeMax) = leftInterval match {
      case hasUpper: WithUpper => (comparableToDouble(hasUpper.upper), hasUpper.includeUpper)
      case _ => (null, true)
    }
    val (rightMin, rightIncludeMin) = rightInterval match {
      case hasLower: WithLower => (comparableToDouble(hasLower.lower), hasLower.includeLower)
      case _ => (null, true)
    }
    val (rightMax, rightIncludeMax) = rightInterval match {
      case hasUpper: WithUpper => (comparableToDouble(hasUpper.upper), hasUpper.includeUpper)
      case _ => (null, true)
    }
    // determine the overlapping degree between predicate interval and column's interval
    val (noOverlap, completeOverlap) = op match {
      // Left < Right or Left <= Right
      // - no overlap:
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap: (If null values exists, we set it to partial overlap.)
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      case LESS_THAN =>
        (
          greaterThanOrEqualTo(leftMin, rightMax),
          if (leftIncludeMax && rightIncludeMin) {
            lessThan(leftMax, rightMin)
          } else {
            lessThanOrEqualTo(leftMax, rightMin)
          })
      case LESS_THAN_OR_EQUAL =>
        (
          if (leftIncludeMin && rightIncludeMax) {
            greaterThan(leftMin, rightMax)
          } else {
            greaterThanOrEqualTo(leftMin, rightMax)
          },
          lessThanOrEqualTo(leftMax, rightMin))
      // Left > Right or Left >= Right
      // - no overlap:
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap: (If null values exists, we set it to partial overlap.)
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      case GREATER_THAN =>
        (
          lessThanOrEqualTo(leftMax, rightMin),
          if (leftIncludeMin && rightIncludeMax) {
            greaterThan(leftMin, rightMax)
          } else {
            greaterThanOrEqualTo(leftMin, rightMax)
          })
      case GREATER_THAN_OR_EQUAL =>
        (
          if (leftIncludeMax && rightIncludeMin) {
            lessThan(leftMax, rightMin)
          } else {
            lessThanOrEqualTo(leftMax, rightMin)
          },
          greaterThanOrEqualTo(leftMin, rightMax))
      // Left = Right
      // - no overlap:
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap:
      //      leftMin            leftMin
      //      rightMin           rightMax
      // --------+------------------+------->
      case EQUALS =>
        // now, min/max/ndv is not null and ndv >= 0
        val isLeftSmallerThanRight = if (leftIncludeMax && rightIncludeMin) {
          leftMax < rightMin
        } else {
          leftMax <= rightMin
        }
        val isLeftBiggerThanRight = if (rightIncludeMax && leftIncludeMin) {
          rightMax < leftMin
        } else {
          rightMax <= leftMin
        }
        (
          isLeftSmallerThanRight || isLeftBiggerThanRight,
          (leftMin == rightMin) && (leftMax == rightMax)
            && (leftIncludeMin == rightIncludeMin)
            && (leftIncludeMax == rightIncludeMax)
            && (leftNdv == rightNdv))
    }
    val allNotNull = (leftNullCount == 0) && (rightNullCount == 0)
    val selectivity = if (noOverlap) {
      0.0
    } else if (completeOverlap && allNotNull) {
      1.0
    } else {
      // For partial overlap, we use an empirical value 1/3 as suggested by the book
      // "Database Systems, the complete book".
      1.0 / 3.0
    }
    Some(selectivity)
  }