[SPARK-20567] Failure to bind when using explode and collect_set in streaming

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 2.2.0
• Fix Version/s: 2.2.0
• Component/s: SQL
• Target Version/s: 2.2.0
• Assignee: Michael Armbrust
• Reporter: Michael Armbrust
• Created: 03/May/17 00:12
• Updated: 03/May/17 05:45
• Resolved: 03/May/17 05:45

Description

Here is a small test case:

  test("count distinct") {
    val inputData = MemoryStream[(Int, Seq[Int])]

    val aggregated =
      inputData.toDF()
        .select($"*", explode($"_2") as 'value)
        .groupBy($"_1")
        .agg(size(collect_set($"value")))
        .as[(Int, Int)]

    testStream(aggregated, Update)(
      AddData(inputData, (1, Seq(1, 2))),
      CheckLastBatch((1, 2))
    )
  }

Issue Links

• links to: [Github] Pull Request #17838 (marmbrus)