[FLINK-31118][table] Add ARRAY_UNION function.

[liuyongvs]

• What is the purpose of the change
  This is an implementation of ARRAY_UNION

• Brief change log
  ARRAY_UNION for Table API and SQL

• Verifying this change
  This change added tests in CollectionFunctionsITCase

• 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): ( no)
  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)

[liuyongvs]

hi @snuyanzin do you have time to review it. and the input args type should do somethinh like calcite array_concat. do you have good idea, because it should be reuse like array_except/array_intersect/

public static final SqlFunction ARRAY_CONCAT = SqlBasicFunction.create(SqlKind.ARRAY_CONCAT, ReturnTypes.LEAST_RESTRICTIVE, OperandTypes.AT_LEAST_ONE_SAME_VARIADIC);

[flinkbot]

CI report:

• d425b1f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[snuyanzin]

From the description it is not clear what happens if only one array is null

[snuyanzin]

Thanks for your contribution.

To be honest i didn't get the logic.
What i did:

1. Built it from the PR's branch
2. Started standalone Flink
3. Via sqlClient submitted several queries

SELECT array_union(array[1], array[2]);
-- result array[1, 2]
-- this is OK

SELECT array_union(array[1], array[2, 3]);
-- result array[1, 2, 3]
--- this is OK

SELECT array_union(array[1], array[2, 3, null]);
-- result array[1, 2, 3, 0]
--- this is NOT OK

SELECT array_union(array[1], array[map['this is a key', 'this is a value']]);
-- result [1, 68]
-- this is NOT OK

SELECT array_union(array[1], array['this is a string']);
-- result [1, 16]
-- this is NOT OK

SELECT array_union(array[1], array[array[1, 2, 3]]);
-- result [1, 24]
-- this is NOT OK

[snuyanzin]

same here

[snuyanzin]

What about ordering?

[liuyongvs]

Thanks for your contribution.

To be honest i didn't get the logic. What i did:

1. Built it from the PR's branch
2. Started standalone Flink
3. Via sqlClient submitted several queries

SELECT array_union(array[1], array[2]);
-- result array[1, 2]
-- this is OK

SELECT array_union(array[1], array[2, 3]);
-- result array[1, 2, 3]
--- this is OK

SELECT array_union(array[1], array[2, 3, null]);
-- result array[1, 2, 3, 0]
--- this is NOT OK

SELECT array_union(array[1], array[map['this is a key', 'this is a value']]);
-- result [1, 68]
-- this is NOT OK

SELECT array_union(array[1], array['this is a string']);
-- result [1, 16]
-- this is NOT OK

SELECT array_union(array[1], array[array[1, 2, 3]]);
-- result [1, 24]
-- this is NOT OK

the result is not ok like SELECT array_union(array[1], array[array[1, 2, 3]]); SELECT array_union(array[1], array['this is a string']); i said before this two array type should be exception. we should do it like ARRAY_CONCAT is calcite. but i do not find a good way to express it in flink. so i ask you do you have a good implements and it is important for other array op like
array_except/array_intersect..

[snuyanzin]

so i ask you do you have a good implements

unfortunately right now i do not have anything where i can just pick it up. You probably could do some research in this direction

[liuyongvs]

hi @snuyanzin the pr i submit before do not think about implicit cast like spark such as ARRAY ARRAY

and i sumit a new pr here #22483
and i close this pr