[SPARK-37458][SS] Remove unnecessary SerializeFromObject from the plan of foreachBatch

[HeartSaVioR]

What changes were proposed in this pull request?

This PR proposes to change the logic on ForeachBatchSink to remove unnecessary SerializeFromObject, via leveraging LogicalRDD instead of ExternalRDD.

Why are the changes needed?

This brings slight performance gain as Spark no longer does unnecessary serde on foreachBatch. In addition, the logic is simpler as we defer the encoding logic to the Dataset.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing UT, and new UTs.

[HeartSaVioR]

The performance gain is captured via custom benchmark code:

HeartSaVioR@222d7a6

Typed

[info] OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
[info] Intel(R) Xeon(R) Platinum 8175M CPU @ 2.50GHz
[info] LogicalRDD vs ExternalRDD:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] LogicalRDD                                          421            437          15         23.8          42.1       1.0X
[info] ExternalRDD                                         851            866          11         11.7          85.1       0.5X

Untyped

[info] OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
[info] Intel(R) Xeon(R) Platinum 8175M CPU @ 2.50GHz
[info] LogicalRDD vs ExternalRDD:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] LogicalRDD                                          469            481           9         21.3          46.9       1.0X
[info] ExternalRDD                                         913            929          13         11.0          91.3       0.5X

The origin DataFrame is typed one and we don't change the type during transformation, so it's not odd untyped case could be slower.

[HeartSaVioR]

I didn't add the benchmark code to this PR since the code is quite specific to capture "before vs after" for this PR. Please let me know if we'd like to add the benchmark in any way.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50085/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50085/

[HeartSaVioR]

cc. @tdas @zsxwing @viirya @xuanyuanking

[SparkQA]

Test build #145613 has finished for PR 34706 at commit af961b2.

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

[HeartSaVioR]

Friendly reminder, @tdas @zsxwing @viirya @xuanyuanking

[dongjoon-hyun]

+1, LGTM.

[dongjoon-hyun]

cc @cloud-fan, too

[HeartSaVioR]

Thanks for reviewing! Let me retrigger builds and merge once build passes.

[HeartSaVioR]

retest this, please

[dongjoon-hyun]

Oops. Sorry, I merged it before seeing your message, @HeartSaVioR .

[dongjoon-hyun]

BTW, thank you so much, @HeartSaVioR and @cloud-fan .

[HeartSaVioR]

Never mind. I guess there's no possibility someone else modifies the relevant code, so we're good to go.
Thanks for taking care of!