[FLINK-37598][table] Support list and map state in PTFs

[twalthr]

What is the purpose of the change

This adds support for list and map state in PTFs via ListView and MapView.

Brief change log

• Update ProcessTableOperator and Runner
• Update docs and JavaDocs

Verifying this change

This change added tests and can be verified as follows:

ProcessTableFunctionTestPrograms#PROCESS_LIST_STATE
ProcessTableFunctionTestPrograms#PROCESS_MAP_STATE

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): yes
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs

[flinkbot]

CI report:

• 4f078ca Azure: FAILURE
• 731559e UNKNOWN

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[snuyanzin]

thank you for addressing feedback

[davidradl]

nit: the code would be cleaner with the if as positive and without the continue:

 if ((stateHandle instanceof ValueState)) {
           final ValueState<RowData> valueState = (ValueState<RowData>) stateHandle;
           final RowData value = valueState.value();
           valueStateToFunction[i] = value;
  }

[twalthr]

I'm following the early return principle here. I think in the end this is a matter of taste. Feel free to code it differently in your PRs.

[davidradl]

I'm following the early return principle here. I think in the end this is a matter of taste. Feel free to code it differently in your PRs.

ok no worries.

[davidradl]

nit: how about rewriting the if and returns as

 if (isDataView) {
        viewType.getChildren().forEach(DataViewUtils::checkForInvalidDataViews);
 }
return isDataView;

[twalthr]

I'm following the early return principle here

[davidradl]

ok no worries

[davidradl]

is it possible to use the equals() method on the functions to assert whether they are equal? If we are not in control as we cannot rely on the equals as they are user written, then I can see the need for a class like the Record Equaliser. I am curious on the use of RecordEqualiser, equaliser implies to me we are looking to make the functions equal - where as we are likely just looking to compare - maybe a RecordComparator might be a better name.

[twalthr]

We are comparing internal data structures here. i.e. RowData. Which might be backed by memory segments. Also SQL semantics need to be considered when comparing nested data. This is why RecordEqualiser and HashFunction are code generated instances.

[davidradl]

@twalthr thanks for addressing the feedback.

[twalthr]

@flinkbot run azure