def caseClassTypeDescriptorImpl[T]()

in scalding-core/src/main/scala/com/twitter/scalding/macros/impl/TypeDescriptorProviderImpl.scala [28:89]
42 lines of code
28 McCabe index (conditional complexity)

  def caseClassTypeDescriptorImpl[T](c: Context)(implicit T: c.WeakTypeTag[T]): c.Expr[TypeDescriptor[T]] =
    caseClassTypeDescriptorCommonImpl(c, false)(T)

  def caseClassTypeDescriptorWithUnknownImpl[T](c: Context)(implicit
      T: c.WeakTypeTag[T]
  ): c.Expr[TypeDescriptor[T]] =
    caseClassTypeDescriptorCommonImpl(c, true)(T)

  /**
   * When flattening a nested structure with Options, the evidentColumn is a column, relative to the the first
   * 0-offset column, that represents evidence of this T, and hence set of columns, are present or absent.
   * This is to handle Option types in text files such as CSV and TSV. a type T is evident if it the
   * evidentColumn.exists
   *
   * primitive numbers are evident case classes are evident if they have at least one evident member.
   *
   * Strings are not evident (we can't distinguish Empty from "") Option[T] is not evident (we can't tell
   * Some(None) from None).
   */
  def evidentColumn(c: Context, allowUnknown: Boolean = false)(tpe: c.universe.Type): Option[Int] = {
    import c.universe._

    def flattenOnce(t: Type): List[Type] =
      t.declarations
        .collect { case m: MethodSymbol if m.isCaseAccessor => m }
        .map(_.returnType.asSeenFrom(t, t.typeSymbol.asClass))
        .toList

    @SuppressWarnings(Array("org.wartremover.warts.OptionPartial"))
    def go(t: Type, offset: Int): (Int, Option[Int]) = {
      val thisColumn = (offset + 1, Some(offset))
      t match {
        case tpe if tpe =:= typeOf[String] =>
          // if we don't allowUnknown here, we treat null and "" is indistinguishable
          // for text formats
          if (allowUnknown) thisColumn
          else (offset + 1, None)
        case tpe if tpe =:= typeOf[Boolean] => thisColumn
        case tpe if tpe =:= typeOf[Short]   => thisColumn
        case tpe if tpe =:= typeOf[Int]     => thisColumn
        case tpe if tpe =:= typeOf[Long]    => thisColumn
        case tpe if tpe =:= typeOf[Float]   => thisColumn
        case tpe if tpe =:= typeOf[Double]  => thisColumn
        // We recurse on Option and case classes
        case tpe if tpe.erasure =:= typeOf[Option[Any]] =>
          val innerTpe = optionInner(c)(tpe).get
          // we have no evidentColumn, but we need to compute the next index
          (go(innerTpe, offset)._1, None)
        case tpe if tpe.typeSymbol.isClass && tpe.typeSymbol.asClass.isCaseClass =>
          val flattened = flattenOnce(tpe)
            .scanLeft((offset, Option.empty[Int])) { case ((off, _), t) => go(t, off) }

          val nextPos = flattened.last._1
          val ev = flattened.collectFirst { case (_, Some(col)) => col }
          (nextPos, ev)
        case _ if allowUnknown => thisColumn
        case t =>
          c.abort(c.enclosingPosition, s"Case class $tpe at $t is not pure primitives or nested case classes")
      }
    }
    go(tpe, 0)._2
  }