[BEAM-10047] Merge the stages 'Gather and Sort' and 'Create Batches'

[nielm]

There is minimal benefit in separating these 2 stages, and significant
benefity in merging them: Gather and Sort encodes incoming
MutationGroups into a List<byte[]> which would contain up to 1GB.
This is then output (copied) to the CreateBatches where it is decoded
back into MutationGroups.

Removing this encode/decode should save up to 2GB of RAM.

Note, this PR is dependent on PR #11528, PR #11532 and PR #11529

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[allenpradeep]

This is great niel. With these changes, there are 3 modes of using SpannerIO write.
a) Use the conventional way(as it was till now) with a grouping factor where data is grouped, sorted, batched and written as per parameters
b) Batching without grouping - Set grouping factor as 1 with a larger batched bytes or cells. This will just ensure data is just batched without sort.
c) No Batching - Set any of the max rows or max mutations or batch bytes to 0 or 1.

Questions:

1. What mode should our import pipeline use? Should it use option b as data in AVRO seems already sorted?
2. Where should we document these modes of operation so that some customer can use these?

[allenpradeep]

I'm good with these changes except the questions I had regarding the usages.
LGTM

[allenpradeep]

Hi Niel,
I see a bunch of unit tests failing on this commit.
I am working on a patch on top of this and i noticed this.

[nielm]

@allenpradeep

1. What mode should our import pipeline use? Should it use option b as data in AVRO seems already sorted?

We can discuss this outside the scope of this PR.

2. Where should we document these modes of operation so that some customer can use these?

I have added a section to the javadoc explaining these 3 modes of operation, and their pros and cons.

[TheNeuralBit]

Retest this please

[nielm]

Retest this please

[TheNeuralBit]

Retest this please

[TheNeuralBit]

Retest this please

[nielm]

Retest this please

[TheNeuralBit]

Retest this please

[udim]

Is this ready to merge?

[tvalentyn]

Run Java PreCommit

[tvalentyn]

[CheckStyle] Attaching ResultAction with ID 'checkstyle' to run 'beam_PreCommit_Java_Commit #11858'.
Setting status of a74866ba56d92d9476006b7e40a0e0ff916748ca to FAILURE with url https://ci-beam.apache.org/job/beam_PreCommit_Java_Commit/11858/ and message: 'Build finished. '
Using context: Java ("Run Java PreCommit")
Finished: ABORTED

Can't tell if tests passed or not, rerunning.

[allenpradeep]

Can we merge this PR? I would want to send out a PR to count bytes written to spanner and that would be dependent on this.

[nielm]

Retest this please

[chamikaramj]

Run Java PostCommit

[chamikaramj]

Thanks. We can merge if post-commit tests pass.