[SPARK-28423][SQL] Merge Scan and Batch/Stream

[cloud-fan]

What changes were proposed in this pull request?

By design, Scan represents a logical data scan, Batch/Stream represents a physical data scan.

However, this doesn't match reality. The logical plan(DataSourceV2Relation) contains Table and the phyiscal plan(BatchScanExec and friends) contains Scan. The operator pushdown happens at planning time, so Scan and Batch/Stream are always created together in the planner rules. That said, Table is the actual logical data scan.

Since there is not much can be separated from Scan and Batch/Stream, almost all the existing DS v2 implementations either implement Scan and Batch/Stream together, or use anonymous class to implement Scan.

In addition, the write side API has no such separation either: it's just WriterBuilder -> BatchWrite/StreamingWrite.

This PR proposes to merge Scan and Batch/Stream, to match the write side API: ScanBuilder -> BatchScan/MicroBatchScan/ContinuousScan.

How was this patch tested?

existing tests

[cloud-fan]

cc @marmbrus @rdblue @jose-torres @gengliangwang

[jose-torres]

This seems fine, but didn’t we decide a while back (Q4 18 I think) not to do it?

[SparkQA]

Test build #107789 has finished for PR 25180 at commit 439d6dc.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

@jose-torres seems there is a misunderstanding. This is on my TODO list a long time ago and I do want to finish it before 3.0.

[SparkQA]

Test build #107794 has finished for PR 25180 at commit af96a60.

• This patch fails to generate documentation.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #107824 has finished for PR 25180 at commit 878eaa5.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[gengliangwang]

retest this please.

[SparkQA]

Test build #107841 has finished for PR 25180 at commit 878eaa5.

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

[dongjoon-hyun]

Could you review this change please, @rdblue ?

[SparkQA]

Test build #107906 has finished for PR 25180 at commit 83f2967.

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

[rdblue]

Sorry I haven't gotten to this yet. I had some unexpected travel and was out of the office. I should be able to take a look at this tomorrow.

[dongjoon-hyun]

Thank you, @rdblue .

[cloud-fan]

@rdblue cool, thanks!

[SparkQA]

Test build #108077 has finished for PR 25180 at commit 9c826f3.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

retest this please

[SparkQA]

Test build #108092 has finished for PR 25180 at commit 9c826f3.

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

[rdblue]

That said, Table is the actual logical data scan.

No, table is a source that can be scanned. A logical scan has filters and a projection.

The operator pushdown happens at planning time, so Scan and Batch/Stream are always created together in the planner rules.

The scan is created in DataSourceV2Strategy, but batch is a lazy field in BatchScanExec. There's no need for the planner strategy to know about the batch and stream objects at all.

This separation would be useful if we decide to move push-down into a batch in the optimizer. We've been discussing options for doing push-down earlier and being able to use stats in the optimizer. If we did that, then the separation between scan and batch/stream would support that. We would introduce a logical node that has a scan that is produced in the optimizer.

[cloud-fan]

I think the separation between Scan and Batch is still useless even if we move the operator pushdown to the optimizer. There is no extra information needed to convert a Scan to a Batch, which means if I have a class that implements Scan, there is no problem for me to implement Batch at the same time.

As a result, almost all the existing DS v2 implementations either implement Scan and Batch/Stream together, or use anonymous class to implement Scan. This makes me believe that we should remove this separation.

Conceptually, the physical scan is represented by InputPartition and PartitionReaderFactory, not the interface that creates them. It makes more sense to use a single interface to represent a logical scan, which creates InputPartition and PartitionReaderFactory.

[cloud-fan]

Hi @jose-torres , can you double-check if my explanation is correct?

[cloud-fan]

Hi @jose-torres , can you double-check if my explanation is correct?

[SparkQA]

Test build #108717 has finished for PR 25180 at commit a7d0c55.

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

[github-actions]

We're closing this PR because it hasn't been updated in a while.
This isn't a judgement on the merit of the PR in any way. It's just
a way of keeping the PR queue manageable.

If you'd like to revive this PR, please reopen it!

[SparkQA]

Test build #115830 has finished for PR 25180 at commit a7d0c55.

• This patch fails build dependency tests.
• This patch does not merge cleanly.
• This patch adds no public classes.