[SPARK-29248][SQL] provider number of partitions when creating v2 data writer factory

[edrevo]

What changes were proposed in this pull request?

When implementing a ScanBuilder, we require the implementor to provide the schema of the data and the number of partitions.

However, when someone is implementing WriteBuilder we only pass them the schema, but not the number of partitions. This is an asymetrical developer experience.

This PR adds a PhysicalWriteInfo interface that is passed to createBatchWriterFactory and createStreamingWriterFactory that adds the number of partitions of the data that is going to be written.

Why are the changes needed?

Passing in the number of partitions on the WriteBuilder would enable data sources to provision their write targets before starting to write. For example:

it could be used to provision a Kafka topic with a specific number of partitions
it could be used to scale a microservice prior to sending the data to it
it could be used to create a DsV2 that sends the data to another spark cluster (currently not possible since the reader wouldn't be able to know the number of partitions)

Does this PR introduce any user-facing change?

No

How was this patch tested?

Tests passed

[edrevo]

@rdblue @cloud-fan , this PR contains the latest feedback from #25990. Sorry for closing the other PR and opening a new one! 😓

[cloud-fan]

ok to test

[SparkQA]

Test build #114068 has finished for PR 26591 at commit 6d9e427.

• 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 #114139 has finished for PR 26591 at commit 6d9e427.

• 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 #114153 has finished for PR 26591 at commit 6d9e427.

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

[cloud-fan]

I keep hitting errors when merging this PR: UnicodeEncodeError: 'ascii' codec can't encode character u'\xe1' in position 71: ordinal not in range(128)

@HyukjinKwon do you have any clue?

[cloud-fan]

I tried to update PR title and description, but has no luck. I'm going to merge it using Github directly.

[cloud-fan]

oh, seems like the email address is invalid

Traceback (most recent call last):
  File "./dev/merge_spark_pr.py", line 577, in <module>
    main()
  File "./dev/merge_spark_pr.py", line 552, in main
    merge_hash = merge_pr(pr_num, target_ref, title, body, pr_repo_desc)
  File "./dev/merge_spark_pr.py", line 147, in merge_pr
    distinct_authors[0])
UnicodeEncodeError: 'ascii' codec can't encode character u'\xe1' in position 71: ordinal not in range(128)

@edrevo can you set up a different email for your git and rebase your commit?

[edrevo]

I think it is the á in my name. I'll change it and rebase it, no problem. The weird thing is, I have previously contributed to spark and the unicode char wasn't a problem back then, so something must have changed in the CI that now breaks with unicode.

[edrevo]

rebased

[SparkQA]

Test build #114239 has finished for PR 26591 at commit 21bbd6b.

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

[cloud-fan]

thanks, merging to master!

[cloud-fan]

@edrevo please open a new PR to add logical write info, thanks!

[edrevo]

will do! many thanks for the patience you've had with me and for steering me in the right direction with the changes.

[rdblue]

@edrevo, @cloud-fan, what is the intended purpose of LogicalWriteInfo?

[cloud-fan]

To make the API more type-safe. I've seen implementations checking if withQueryId is called and called only once. With LogicalWriteInfo interface, people can just get info from it and don't need to worry about potential mistakes at Spark side.

We can discuss more after @edrevo open the PR.

[cloud-fan]

A real example: when @edrevo adding the numPartitions info using withNumPartitions, none of us realizing that it's not set at the streaming side. When @edrevo refactoring the code using LogicalWriteInfo, we realize it immediately that it's missing at the streaming side and then we propose PhyiscalWriteInfo.