private def parameterClassEquals()

in flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala [760:796]
35 lines of code
47 McCabe index (conditional complexity)

  private def parameterClassEquals(candidate: Class[_], expected: Class[_]): Boolean =
    candidate == null ||
      candidate == expected ||
      expected == classOf[Object] ||
      candidate == classOf[Object] || // Special case when we don't know the type
      expected.isPrimitive && Primitives.wrap(expected) == candidate ||
      candidate == classOf[Date] && (expected == classOf[Int] || expected == classOf[JInt]) ||
      candidate == classOf[Time] && (expected == classOf[Int] || expected == classOf[JInt]) ||
      candidate == classOf[StringData] && expected == classOf[String] ||
      candidate == classOf[String] && expected == classOf[StringData] ||
      candidate == classOf[TimestampData] && expected == classOf[LocalDateTime] ||
      candidate == classOf[Timestamp] && expected == classOf[TimestampData] ||
      candidate == classOf[TimestampData] && expected == classOf[Timestamp] ||
      candidate == classOf[LocalDateTime] && expected == classOf[TimestampData] ||
      candidate == classOf[TimestampData] && expected == classOf[Instant] ||
      candidate == classOf[Instant] && expected == classOf[TimestampData] ||
      classOf[RowData].isAssignableFrom(candidate) && expected == classOf[Row] ||
      candidate == classOf[Row] && classOf[RowData].isAssignableFrom(expected) ||
      classOf[RowData].isAssignableFrom(candidate) && expected == classOf[RowData] ||
      candidate == classOf[RowData] && classOf[RowData].isAssignableFrom(expected) ||
      candidate == classOf[DecimalData] && expected == classOf[BigDecimal] ||
      candidate == classOf[BigDecimal] && expected == classOf[DecimalData] ||
      (candidate.isArray &&
        expected.isArray &&
        candidate.getComponentType.isInstanceOf[Object] &&
        expected.getComponentType == classOf[Object])

  private def parameterDataTypeEquals(internal: LogicalType, parameterType: DataType): Boolean = {
    val paraInternalType = fromDataTypeToLogicalType(parameterType)
    if (isRaw(internal) && isRaw(paraInternalType)) {
      getDefaultExternalClassForType(internal) == getDefaultExternalClassForType(paraInternalType)
    } else {
      // There is a special equal to GenericType. We need rewrite type extract to RowData etc...
      paraInternalType == internal ||
      toInternalConversionClass(internal) == toInternalConversionClass(paraInternalType)
    }
  }