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.

This closes #14437

[FLINK-20693][table-planner-blink][python] Port BatchExecPythonCorrelate and StreamExecPythonCorrelate to Java

This closes #14503.

The DeclarativeSlotPool.registerNewSlotsListener can be used to register a
DeclarativeSlotPool.NewSlotsListener which is notified whenever the
DeclarativeSlotPool receives new slots or reserved slots are freed.

This closes #14482.

This closes #14518.

It was not possible to just run 'mvn spotless:check', but instead one needed to use the fully qualified name of the plugin, i.e., 'mvn com.diffplug.spotless:spotless-maven-plugin:check'.

This is due to 'force-shading'; this module is declared as a child of the flink-parent pom (i.e., the root pom of the project), but is not actually declaring that very module as it's parent.
This seems to throw of maven, which fails to retrieve the plugin information from the parent.

Note that this is just a band-aid; a proper fix would declare flink-parent as the parent of force-shading, but this is the easier fix.

[FLINK-17331][runtime] Explicitly get the ByteBuf length of all classes which is written to NettyMessage

Currently, the length of some header fields in NettyMessage is hardcoded: InputChannelID, ExecutionAttemptID, e.t.c.
So if we make some changes for such field, then we are not ware that it also needs to change the respective length for related netty messages component.
This PR explicitly get the ByteBuf length of all classes which is written to NettyMessage to avoid such problems.

Display more informative message stating that Watermarks are only available when using
event time if there are no watermarks to display in the web UI.

This closes #14483.

This closes #14514.

This file can be used via:

$ git config blame.ignoreRevsFile .git-blame-ignore-revs