[FLINK-9601][state]Try to reuse the snapshotData array as the partitioned destination only when its capacity is enough

[sihuazhou]

What is the purpose of the change

In CopyOnWriteStateTableSnapshot, we only reuse the snapshotData as the partitioned destination array when it's capacity enough is enough.

Brief change log

• Try to reuse the snapshotData array as the partitioned destination on when it's capacity is enough.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
• The S3 file system connector: (no)

Documentation

No

[sihuazhou]

CC @StefanRRichter

[StefanRRichter]

Thanks for the contribution. I think the general idea is good, but I think the implementation can be improved. How about changing CopyOnWriteStateTable::snapshotTableArrays() in such ways that the returned snapshot array is already created with a size of max(whateverTheCurrentCodeDoes, size()). This prevents that we ever have to create another array, in particular in this corner case where we are dealing with huge arrays. Then the remaining code should automagically work without further changes. Adding some nice comment might help that our optimized snapshotting algorithm assumes that the array can hold the flattened entries. What do you think?

[sihuazhou]

@StefanRRichter Thanks for the review, I think that makes a lot of sense, will change it according to your suggestion.

[StefanRRichter]

LGTM now 👍 Will merge.