By moving the shut down of checkpoint services out of the CheckpointCoordinator,
it is now possible to reuse these services across different CheckpointCoordinators.

This closes #14553.

[hotfix][tests] Replace explicit ExecutionGraphBuilder.buildGraph calls with TestingExecutionGraphBuilder

[FLINK-20846] Factor out CompletedCheckpointStore and CheckpointIDCounter creation in ExecutionGraphBuilder.buildGraph

Apply spotless formatting

This closes #14425.

This adds a name-based field mode to the Row class. A row can
operate in 3 different modes: name-based, position-based, or
a hybrid of both when leaving the Flink runtime. It simplifies
the handling of large rows (possibly with hundreds of fields)
and will make it easier to switch between DataStream API and
Table API.

See the documentation of the Row class for more information.

This closes #14420.

This fixes the ClassNotFound problem when create a generic but not loaded function in Hive catalog

This closes #14534

In order to maintain the invariant that a failed createPartitionRequestClient won't leave
a failed connection future in the clients field, this commit removes the client future
if its creation fails.

This closes #14528.

Co-authored-by: Nshuo.cs <shuo.cs@alibaba-inc.com>

Co-authored-by: Nzhuxiaoshang <zhushang@qutoutiao.net>

This closes #14524.

This closes #14527

This closes #14464

Some of the rel data types for legacy planner:
- GenericRelDataType
- ArrayRelDataType
- MultisetRelDataType

Do not implement the digest correctly, especially for
GenericRelDataType , the RelDataTypeFactory caches the type instances
based on its digest, a wrong digest impl would mess up the type instance
creation, e.g. without this patch, all the GenericRelDataType instances
have same digest `ANY`.

The MapRelDataType is not fixed because we ignore the key/value
nullability when creating the type, which conflicts with the digest.
There is already issue (FLINK-20785) to track this.

This closes #14451

This closes #14521.

[FLINK-20594][runtime] Replaced DefaultExecutionSlotAllocator and its Factory class by SlotSharingExecutionSlotAllocator and its Factory.

TestingPhysicalSlotProvider was moved into its own class file and generalized to match other testcases besides SlotSharingExecutionSlotAllocatorTest.

This closes #14465.

[FLINK-18654][docs][jdbc] Correct missleading documentation in "Partitioned Scan" section of JDBC connector

This closes #14523

[FLINK-20606][hive][table] Fix CREATE FUNCTION is failed under Hive catalog in SQL CLI using -j option

This closes #14392

This reduces the number of failures in UnalignedCheckpointITCase.union to ~2/100.

In scenarios with multiple inputs (e.g. co-group; not union) one input may receive a
checkpoint barrier while the second input is still restoring state. This (previous)
state is currently not included into the snapshot, which therefore will be incomplete.

In StreamTaskNetworkInput.prepareSnapshot, internal channelIndex
is inconsistent with CheckpointedInputGate channelIndex.

This change fixes data loss in UnalignedCheckpointITCase.cogroup.

Most of the bugs in UC are revealed with higher back-pressure which is not created with alignment timeout.
This change disables it by default and adds a new test (p=20) with the timeout enabled.