title: Spark SQL UDF with complex input parameter
body: <p>I'm trying to use UDF with input type Array of struct.
  I have the following structure of data this is only relevant part of a bigger structure</p>
  
  <pre><code>|--investments: array (nullable = true)
      |    |-- element: struct (containsNull = true)
      |    |    |-- funding_round: struct (nullable = true)
      |    |    |    |-- company: struct (nullable = true)
      |    |    |    |    |-- name: string (nullable = true)
      |    |    |    |    |-- permalink: string (nullable = true)
      |    |    |    |-- funded_day: long (nullable = true)
      |    |    |    |-- funded_month: long (nullable = true)
      |    |    |    |-- funded_year: long (nullable = true)
      |    |    |    |-- raised_amount: long (nullable = true)
      |    |    |    |-- raised_currency_code: string (nullable = true)
      |    |    |    |-- round_code: string (nullable = true)
      |    |    |    |-- source_description: string (nullable = true)
      |    |    |    |-- source_url: string (nullable = true)
  </code></pre>
  
  <p>I declared case classes:</p>
  
  <pre class="lang-scala prettyprint-override"><code>case class Company(name: String, permalink: String)
  case class FundingRound(company: Company, funded_day: Long, funded_month: Long, funded_year: Long, raised_amount: Long, raised_currency_code: String, round_code: String, source_description: String, source_url: String)
  case class Investments(funding_round: FundingRound)
  </code></pre>
  
  <p>UDF declaration:</p>
  
  <pre class="lang-scala prettyprint-override"><code>sqlContext.udf.register("total_funding", (investments:Seq[Investments])  =&gt; {
       val totals = investments.map(r =&gt; r.funding_round.raised_amount)
       totals.sum
  })
  </code></pre>
  
  <p>When I'm executing the following transformation the result is as expected</p>
  
  <pre class="lang-scala prettyprint-override"><code>scala&gt; sqlContext.sql("""select total_funding(investments) from companies""")
  res11: org.apache.spark.sql.DataFrame = [_c0: bigint]
  </code></pre>
  
  <p>But when an action executed like collect I have an error:</p>
  
  <pre><code>Executor: Exception in task 0.0 in stage 4.0 (TID 10)
  java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to $line33.$read$$iwC$$iwC$Investments
  </code></pre>
  
  <p>Thank you for any help.</p>