[SPARK-29999][SS] Handle FileStreamSink metadata correctly for empty partition

[HeartSaVioR]

What changes were proposed in this pull request?

This patch checks the existence of output file for each task while committing the task, so that it doesn't throw FileNotFoundException while creating SinkFileStatus. The check is newly required for DSv2 implementation of FileStreamSink, as it is changed to create the output file lazily (as an improvement).

JSON writer for example:

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonOutputWriter.scala

Lines 49 to 60 in 9ec2a4e

	override def write(row: InternalRow): Unit = {
	val gen = jacksonGenerator.getOrElse {
	val os = CodecStreams.createOutputStreamWriter(context, new Path(path), encoding)
	// create the Generator without separator inserted between 2 records
	val newGen = new JacksonGenerator(dataSchema, os, options)
	jacksonGenerator = Some(newGen)
	newGen
	}
	gen.write(row)
	gen.writeLineEnding()
	}

Why are the changes needed?

Without this patch, FileStreamSink throws FileNotFoundException when writing empty partition.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added UT.

[HeartSaVioR]

cc. @gengliangwang as he led DSv2 implementation for FileDataSource

[SparkQA]

Test build #114293 has finished for PR 26639 at commit fd276b5.

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

[dongjoon-hyun]

+1, LGTM. Merged to master.
Thank you so much, @HeartSaVioR !

[dongjoon-hyun]

cc @zsxwing and @gatorsmile

[HeartSaVioR]

Thanks for reviewing and merging!

[gatorsmile]

This only affect DSV2? How about DSV1?

[gengliangwang]

I don't think it is DSV2 only.
It should be caused by this PR: #23052
Checking file existence in each task commit is kind of expensive. We can have a better fix.

[dongjoon-hyun]

Please create a follow-up for DSv1 with the similar test coverage~

[HeartSaVioR]

@gengliangwang Thanks for the pointer on root issue. Missed it. Based on the root issue, this issue wouldn't only reside on DSv2.

In fact, the fix takes the easiest and non-intrusive approach since the issue is only found from streaming sink. There's another possible fix if we are not happy with that; let commitTask (optionally) receives actual list of files being written instead of dealing with its own tracking list. That would change the interface of FileCommitProtocol. (FileFormatDataWriter.commit() can pass the list as it tracks with statsTrackers, but SparkHadoopWriter doesn't do it, so the list would need to be optional, and we still need to check the existence if there's no hint.)

Would it make sense? I feel the change is bigger than just a follow-up issue, as we would change the interface.

@gatorsmile That's not only DSv2 issue but also DSv1 issue; I guess we'd be better to move the new UT to FileStreamSinkSuite so that the UT covers both. WDYT? If it works for you, I'll craft a followup PR. Thanks!

[HeartSaVioR]

@dongjoon-hyun Sorry I saw your comment just after commenting mine. I'll craft a follow-up PR to move the UT to base class so that it could test both DSv1 and DSv2. Thanks!

[zsxwing]

I think the empty partition optimization seems overkill if every task needs to check file existence for each file. We should fix the root issue and revert this file sink fix. The unit test is great and it's worth to keep it to avoid regressions like this.

[dongjoon-hyun]

I'm also fine with the reverting #23052 if we keep these unit tests like @zsxwing 's advice.

The unit test is great and it's worth to keep it to avoid regressions like this.

[HeartSaVioR]

I'm OK with reverting both #23052 and this, and let me come up with another PR (maybe minor PR or reopening the JIRA issue and change the description) which only contains the UT.

I guess it's still viable to try to fix the root issue, like the proposal I proposed in #26639 (comment) as @zsxwing 's comment applies only when #23052 requires this patch. But for completeness, even we would want to keep #23052, the ideal step would be 1) reverting both -> 2) adding UT -> 3) submitting revised version of #23052 which contains a good approach for streaming sink as well.

[gengliangwang]

@HeartSaVioR Yes please revert both and keep the UT in this PR(maybe more UTs).

[HeartSaVioR]

Could I get some guidance here for revert? Because I have seen both approaches in community for reverting: 1) PR (I can also handle) 2) direct (I can't handle, only possible from committers) and given we would like to revert two issues which I couldn't find some reference. If we prefer reverting via PR, would it be OK to have a PR which contain two revert commits? (title would be a bit longer to contain both)

[gengliangwang]

Yes, I think a PR containing two revert commits is OK