[SPARK-38647][SQL] Add SupportsReportOrdering mix in interface for Scan (DataSourceV2) by EnricoMi · Pull Request #35965 · apache/spark

EnricoMi · 2022-03-24T15:55:29Z

What changes were proposed in this pull request?

As SupportsReportPartitioning allows implementations of Scan provide Spark with information about the exiting partitioning of data read by a DataSourceV2, a similar mix in interface SupportsReportOrdering should provide order information.

Why are the changes needed?

This prevents Spark from sorting data if they already exhibit a certain order provided by the source.

Does this PR introduce any user-facing change?

It adds SupportsReportOrdering mix in interface.

How was this patch tested?

This adds tests to DataSourceV2Suite, similar to the test for SupportsReportPartitioning.

AmplabJenkins · 2022-03-24T18:55:34Z

Can one of the admins verify this patch?

EnricoMi · 2022-04-12T14:27:31Z

@sunchao @dongjoon-hyun looks like #35657 (20ffbf7) broke reusing partitioning and order provided by DS V2 in my tests, which worked before rebase. The query planner now introduces a shuffle and sort again.

Any suggestion how Spark can pick up the existing partitioning and order with the new partitioning design? Any plans for DS V2 to report ordering?

EnricoMi · 2022-04-12T14:33:44Z

I have found the issue, option spark.sql.sources.v2.bucketing.enabled moved to default false, so I have to enable that to have order and partitioning being reused.

EnricoMi · 2022-04-12T14:53:04Z

@sunchao #35657 says "have V2 data sources to report distribution and ordering to Spark on read path": How do V2 sources report their ordering?

sunchao · 2022-04-12T20:21:22Z

@EnricoMi the ordering is not supported yet, but potentially can be done by adding another V2 RangePartitioning which coins the catalyst RangePartitioning class to report V2 ordering spec.

cc @cloud-fan @c21

EnricoMi · 2022-04-12T21:42:06Z

So you are saying ordering is seen as being part of partitioning. This kind of makes sense.

Are you foreseeing that partitioning can transport in-partition order, rather than global order? For instance, some hash partitioning (having no global order) with some in-partition order as used by KeyValueGroupedDataset?

sunchao · 2022-04-13T15:56:57Z

Yes I think it could be a useful addition to KeyGroupedPartitioning, for instance to skip the sort phase in SortMergeJoin. This property may not be preserved though, if there are multiple input partitions sharing the same partition values, as Spark needs to group them.

cloud-fan · 2022-04-14T08:51:41Z

The internal SparkPlan.outputOrdering property is for data ordering within partitions. If a data source scan want to report a global ordering, it needs to report RangePartitioning as well as a data ordering within partitions using the same keys.

That said, I think a new API SupportsReportOrdering makes sense to report the data ordering within partitions.

EnricoMi · 2022-04-19T14:17:13Z

@cloud-fan thanks for clarification, I have incorporated this distinction into the documentation of SupportsReportOrdering.

EnricoMi · 2022-04-26T20:26:25Z

@cloud-fan who would be best to review this?

HyukjinKwon · 2022-04-29T03:41:24Z

cc @aokolnychyi too FYI

sunchao · 2022-04-29T17:04:21Z

This is not correct since Spark can combine input partitions (see groupPartitions).

Combining the partitions through groupPartitions does not preserve order (when concatenated), that is right. It could be preserved by merging the partitions, though.
The outputOrdering of the data source is as stated. Whoever calls into groupPartitions should consider that order is not preserved (depending on how the Seq[InputPartition] is used) if existing order is desired.

Whoever calls into groupPartitions should consider ...
Yes, although groupPartitions is called by Spark right now so we should update accordingly in this PR too. Otherwise, after this PR the outputOrdering could be incorrect if grouping happen.

Alright, when partition grouping happens and multiple partitions are grouped, ordering cannot be preserved.
I have modified outputOrdering not to return the ordering in that case. Making grouped partitions preserve order is out-of-scope of this PR.

sunchao · 2022-05-03T05:21:57Z

Hmm when partitions.length == 1 I think partition combining can still happen? partitions is of type Seq[Seq[InputPartition]] so even if its length = 1, the inner Seq[InputPartition] can still contain more than one elements.

You are right, I mis-interpreted if (partitions.length == 1) SinglePartition in def outputPartitioning, but partitions already incorporates groupedPartitions. Fixing that.

sunchao · 2022-05-03T05:26:32Z

I'm not sure if logicalLink is the idiomatic way of processing a logical plan here (don't see it used much elsewhere) - normally we do in an optimizer rule like V2ScanPartitioning or V2Writes.
@cloud-fan @viirya do you see any issue with this?

I'm not totally sure this is the correct approach. Could we follow the existing pattern and add an optimizer rule to populate this? we can re-use existing V2ScanPartitioning too (and change its name).

Awesome! Rules is where the magic happens...
Integrating this into V2ScanPartitioning makes this so much nicer, gets rid of the awful this.logicalLink.isDefined.
I have renamed V2ScanPartitioning to V2ScanPartitioningAndOrdering.

sunchao · 2022-05-03T16:33:27Z

nit: spaces

sunchao · 2022-05-03T16:37:16Z

This may need to be updated since column "i" may not be in the "partitionKeys".

EnricoMi · 2022-05-09T11:26:48Z

@sunchao I have pushed all my changes

EnricoMi · 2022-05-16T10:18:28Z

@sunchao @HyukjinKwon @aokolnychyi @cloud-fan I have addressed comments and rebased.

sunchao

Sorry to get back so late @EnricoMi . Just got time to go through this PR again. Some more comments: mostly nits except one.

sunchao · 2022-05-20T23:34:57Z

nit: indentation looks off here

sunchao · 2022-05-20T23:37:48Z

nit: space before Arrays

sunchao · 2022-05-20T23:42:24Z

nit: break this into multiple lines:
MyNamedReference(String part) { this.parts = new String[] { part }; }

similarly for other places

sunchao · 2022-05-20T23:42:35Z

sunchao · 2022-05-20T23:45:05Z

nit: indent

sunchao · 2022-05-20T23:54:13Z

I'm not totally sure this is the correct approach. Could we follow the existing pattern and add an optimizer rule to populate this? we can re-use existing V2ScanPartitioning too (and change its name).

EnricoMi · 2022-06-09T18:33:33Z

@sunchao @cloud-fan all comments addressed, all tests green

sunchao

LGTM. Thanks, and sorry for the late response.

sunchao · 2022-06-17T22:35:16Z

nit nit: can we keep this one-line?
valscanRules=Seq[LogicalPlan=>LogicalPlan](partitioning, ordering)

Sure, done. Thanks for the review!

…sist

…tionAwareDataSource

sunchao · 2022-06-21T17:40:36Z

Merged to master, thanks!

EnricoMi · 2022-06-21T17:43:54Z

Thanks @sunchao

github-actionsBot added the SQL label Mar 24, 2022

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch 2 times, most recently from 2efc6ef to 818e0f7Compare March 25, 2022 16:00

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch 3 times, most recently from 7c67284 to 33623c2Compare April 12, 2022 14:22

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch from 36e2678 to 699c607Compare April 19, 2022 14:11

sunchao reviewed Apr 29, 2022
View reviewed changes

sunchao reviewed May 3, 2022
View reviewed changes

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch from 1b024e8 to f421607Compare May 9, 2022 11:20

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch from f421607 to 4ba19f7Compare May 16, 2022 10:16

sunchao reviewed May 20, 2022
View reviewed changes

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch 3 times, most recently from 7c90fd1 to 2e1f5d9Compare June 9, 2022 08:16

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch from 2e1f5d9 to 73c13deCompare June 16, 2022 08:15

sunchao approved these changes Jun 17, 2022
View reviewed changes

EnricoMi added 22 commits June 18, 2022 10:31

Add output ordering to DataSourceV2
35b693c

Fix line length in comment
cb0394f

Add ordering and partitioning tests
0c7ded1

Add test java data source
56e0633

Fix lint errors, remove redundant class
0fa5478

Fix error after rebase master
84cf760

Use new V2ExpressionUtils.toCatalystOrdering
f89330b

Set spark.sql.sources.v2.bucketing.enabled to true
e6a866e

Reworked tests
b4e95d0

Parametrize test to remove defs and make scala style happy
132ea73

Fix scalastyle issues
1408322

Fix Java lint errors
6fcf10a

Make tests work with Scala 2.13
ec332b7

Revert whitespace changes in unrelated areas
929b583

When multiple partitions are grouped, partition ordering does not per…
eed8370
…sist

SinglePartition could also group multiple input partitions
275572e

Fix spaces in for loop clause
5c42977

Reuse JavaPartitionAwareDataSource.MyScanBuilder in JavaOrderAndParti…
0eefc21
…tionAwareDataSource

Return actual partition keys in Java test data source
f14a1ff

Fix formatting according to code review
2078edd

Move extraction of scan ordering into rule
3ebf184

Define scanRules in one line
e444a5d

EnricoMi force-pushed the branch-datasourcev2-output-ordering branch from 73c13de to e444a5dCompare June 18, 2022 08:31

sunchao closed this in b588d07Jun 21, 2022

EnricoMi deleted the branch-datasourcev2-output-ordering branch June 21, 2022 17:44

Uh oh!

Conversation

EnricoMi commented Mar 24, 2022

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

AmplabJenkins commented Mar 24, 2022

Uh oh!

EnricoMi commented Apr 12, 2022

Uh oh!

EnricoMi commented Apr 12, 2022

Uh oh!

EnricoMi commented Apr 12, 2022

Uh oh!

sunchao commented Apr 12, 2022

Uh oh!

EnricoMi commented Apr 12, 2022

Uh oh!

sunchao commented Apr 13, 2022

Uh oh!

cloud-fan commented Apr 14, 2022

Uh oh!

EnricoMi commented Apr 19, 2022

Uh oh!

EnricoMi commented Apr 26, 2022

Uh oh!

HyukjinKwon commented Apr 29, 2022

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

EnricoMi commented May 9, 2022

Uh oh!

EnricoMi commented May 16, 2022

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!