[SPARK-14257][SQL]Allow multiple continuous queries to be started from the same DataFrame

[zsxwing]

What changes were proposed in this pull request?

Make StreamingRelation store the closure to create the source in StreamExecution so that we can start multiple continuous queries from the same DataFrame.

How was this patch tested?

test("DataFrame reuse")

[SparkQA]

Test build #54472 has finished for PR 12049 at commit 50c39b8.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(

[marmbrus]

I think its confusing to have an opaque function that sometimes is creating a source and sometimes returning a static source. Its not going to be clear from explain() which mode you are in.

[zsxwing]

I think its confusing to have an opaque function that sometimes is creating a source and sometimes returning a static source. Its not going to be clear from explain() which mode you are in.

How about adding a new Relation for a static source (maybe call it StreamExecutionRelation)?

[marmbrus]

Or StreamRelation could just hold a DataSource and we could have a Map[DataSource, Source] here thats initialized at startup.

[zsxwing]

I tried Map[DataSource, Source] but failed because of RichSource.

  implicit class RichSource(s: Source) {
    def toDF(): DataFrame = Dataset.ofRows(sqlContext, StreamingRelation(s))

    def toDS[A: Encoder](): Dataset[A] = Dataset(sqlContext, StreamingRelation(s))
  }

If we only have StreamingRelaction(DataSource), then RichSource needs to create a DataSource for Source dynamically.

So the above codes will be changed to

  implicit class RichSource(s: Source) {
    def toDF(): DataFrame = Dataset.ofRows(sqlContext, StreamingRelation(DataSource(sqlContext, className = ...)))

    def toDS[A: Encoder](): Dataset[A] = Dataset(sqlContext, StreamingRelation(sqlContext, className = ...))
  }

Here I don't know what to fill for className. Without code generation, we won't be able to create a new class for different Source instances. This seems too complicated.

Therefore, I used the StreamExecutionRelation idea finally.

[zsxwing]

Its not going to be clear from explain() which mode you are in.

By the way, the stream DataFrame has not yet supported explain. Should we fix it now?

[SparkQA]

Test build #54653 has finished for PR 12049 at commit 6196790.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(dataSource: DataSource, output: Seq[Attribute]) extends LeafNode
  • case class StreamingExecutionRelation(source: Source, output: Seq[Attribute]) extends LeafNode

[SparkQA]

Test build #2722 has finished for PR 12049 at commit 6196790.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(dataSource: DataSource, output: Seq[Attribute]) extends LeafNode
  • case class StreamingExecutionRelation(source: Source, output: Seq[Attribute]) extends LeafNode

[SparkQA]

Test build #2723 has finished for PR 12049 at commit 6196790.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(dataSource: DataSource, output: Seq[Attribute]) extends LeafNode
  • case class StreamingExecutionRelation(source: Source, output: Seq[Attribute]) extends LeafNode

[SparkQA]

Test build #54878 has finished for PR 12049 at commit aa55afe.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #54879 has finished for PR 12049 at commit ac51850.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #54881 has finished for PR 12049 at commit 9b5f007.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[marmbrus]

nit: i don't see anything named _logicalPlan

[marmbrus]

Maybe include a description of how this gets turned into a StreamingExecutionRelation and who's responsibility that is.

[zsxwing]

Done

[SparkQA]

Test build #54932 has finished for PR 12049 at commit 527f55f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #54955 has finished for PR 12049 at commit 48d760e.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(dataSource: DataSource, sourceName: String, output: Seq[Attribute])

[zsxwing]

retest this please

[SparkQA]

Test build #54994 has finished for PR 12049 at commit 48d760e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class StreamingRelation(dataSource: DataSource, sourceName: String, output: Seq[Attribute])

[marmbrus]

Thanks, merging to master.