[FLINK-21949][table] Support ARRAY_AGG aggregate function #23411

Jiabao-Sun · 2023-09-13T12:58:34Z

What is the purpose of the change

[FLINK-21949][table] Support ARRAY_AGG aggregate function

Some nosql databases like mongodb and elasticsearch support nested data types.
Aggregating multiple rows into ARRAY is a common requirement.

Brief change log

Introduce built in function ARRAY_AGG([ ALL | DISTINCT ] expression) to return an array that concatenates the input rows
and returns NULL if there are no input rows. NULL values will be ignored. Use DISTINCT for one unique instance of each value.

SELECT ARRAY_AGG(f1)
  FROM tmp
 GROUP BY f0

Note that we have made some simplifications based on Calcite's SqlLibraryOperators.ARRAY_AGG.

-- calcite
ARRAY_AGG([ ALL | DISTINCT ] value [ RESPECT NULLS | IGNORE NULLS ] [ ORDER BY orderItem [, orderItem ]* ] )
-- flink
ARRAY_AGG([ ALL | DISTINCT ] expression)

The differences from Calcite are as follows:

Null values are ignored.
The order by expression within the function is not supported because the complete row record cannot be accessed within the function implementation.
The function returns null when there's no input rows, but calcite definition returns an empty array. The behavior was referenced from BigQuery and Postgres.

Verifying this change

ITCase and UnitCase are added.

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)

flinkbot · 2023-09-13T13:05:42Z

CI report:

d898519 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Jiabao-Sun · 2023-10-10T06:50:48Z

Hi @wuchong, could you help review this when you have time?

snuyanzin

thanks for the contribution and sorry for the delay

one thing that I noticed: during aggregation there is java's hashmap is used
unfortunately java comparison rules are not fully in sync with SQL comparison rules
for that reason we can not use hash mpas here for generic cases.

We need to use custom expression evaluator, example could be seen in ArrayDistinct implementation

Jiabao-Sun · 2023-12-07T13:57:11Z

Thanks @snuyanzin for the review.
Sorry, I was a little puzzled because I did not use HashMap in Arrayaggfunction.
Could you help pinpoint exactly where it is?

Jiabao-Sun · 2023-12-07T14:46:38Z

Do you mean using equalityEvaluator to compare when merging and retracting lists?

snuyanzin · 2023-12-07T15:14:02Z

yes, sorry was not clear enough

to compare when merging and retracting lists?

yes

Jiabao-Sun · 2023-12-18T08:31:46Z

@flinkbot run azure

Jiabao-Sun · 2023-12-19T09:56:22Z

Hi @snuyanzin, could you help take a look again when you have time?
Thanks.

dawidwys · 2024-01-15T15:38:43Z

#23411 (review)

It took me a while to understand that myself, but I think it's actually ok to depend on the Object#equals/hashcode in AggregateFunctions @snuyanzin what you said is correct for ArrayDistinctFunction, because:

it is a scalar function, so it can be chained with other functions without an exchange before it
It gets data from two input(s) (previous operator and e.g. a literal) haystack and needle may come from different operators which may produce data in different formats e.g. GenericRowData and BinaryRowData

In AggregateFunctions we will always have all records as BinaryRowData (and alike) and those equals/hashcode should work just fine. (It may be a different story when we support STRUCTURED_TYPE#equals/hashcode, but we will need to revisit most of the operators then, because all MapView(s) will work incorrectly). This is what we do in other aggregate functions already. Take a look at e.g. JsonArrayAggFunction or FirstValueWithRetractAggFunction

I haven't seen the previous version @Jiabao-Sun , but I believe we can remove the equalityHandler and make the ArrayAggFunction look similar to JsonArrayAggFunction.

dawidwys · 2024-01-15T15:40:40Z

.../src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlAggFunction.java

    private final List<RelDataType> paramTypes;

    private BridgingSqlAggFunction(
+            FlinkContext context,


What is the reason for those changes? Is it only to be able to create the equalityHandler? If so, let's keep it simple and remove those.

Yes, I'll revert that changes.
Thanks.

snuyanzin · 2024-01-15T23:26:00Z

thanks a lot for the explanation @dawidwys you're right

snuyanzin · 2024-01-15T23:27:11Z

...e-planner/src/test/java/org/apache/flink/table/planner/functions/ArrayAggFunctionITCase.java

+                                Arrays.asList(
+                                        Row.of("A", new Integer[] {1, 2}),
+                                        Row.of("B", new Integer[] {2, 2, 3}),
+                                        Row.of("C", new Integer[] {3}),


I wonder why here is just

Row.of("C", new Integer[] {3}),

based on input I would expect

Row.of("C", new Integer[] {3, null}),

or did I miss anything?

Also similar query for postgres

with input(a, c) as ( select 'a', 1 union all select 'a', 2 union all select 'c', 3 union all select 'c', null ) select a, array_agg(distinct c) from input group by a

gives

a|array_agg| -+---------+ a|{1,2} | c|{3,NULL} |

-- calcite ARRAY_AGG([ ALL | DISTINCT ] value [ RESPECT NULLS | IGNORE NULLS ] [ ORDER BY orderItem [, orderItem ]* ] ) -- flink ARRAY_AGG([ ALL | DISTINCT ] expression)

This function simplifies some aspects compared to calcite.

Currently, the Flink parser does not support parsing RESPECT NULLS | IGNORE NULLS, but it is still possible to make changes to support it.

AggregateUtil#extractDistinctInformation 848~858 ignores the ignoreNulls fields.

flink/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala

Lines 848 to 858 in 6bdb4f7

AggregateCall.create(

call.getAggregation,

false,

false,

false,

call.getArgList,

-1, // remove filterArg

null,

RelCollations.EMPTY,

call.getType,

call.getName)

ListView does not support null values.

flink/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/ListView.java

Lines 45 to 47 in 40cbf23

* <p>Note: Elements of a {@link ListView} must not be null. For heap-based state backends, {@code

* hashCode/equals} of the original (i.e. external) class are used. However, the serialization

* format will use internal data structures.

For that reasons I made some simplification. If we need it, I can make some attempts.

It's ok to support it only partially however the main concern here:
I would expect more consistent result
since there is also input Row.ofKind(INSERT, "D", null), and expected value Row.of("D", null),
So I would expect for ths either containing nulls or both not containing null however not mixed

UPD: to be more clear:
after playing with Postgres and BigQuery I noticed that both respect nulls by default and for BigQuery to make it ignoring nulls it should be specified explicitely. Is there a reason why for Flink it is done differently? Probably it's better to have similar behaviour

I don't tell that we need to support RESPECT NULLS | IGNORE NULLS syntax, however need to make RESPECT NULLS default behaviour to be on same page with vendors

Thanks @snuyanzin, I'm trying to support RESPECT NULLS | IGNORE NULLS syntax.

@snuyanzin
ListView<T> cannot hold null values, and since it is a generic List, it cannot be represented by a specific object for null values. Is there any good solution for this?

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/JsonArrayAggFunction.java

Line 65 in d92ab39

private static final StringData NULL_STR = StringData.fromString("null");

I think it's a very good point. It would be nice to be compatible with SQL and other vendors.

One idea is we could keep the null indices and set those while retrieving the result. Happy to hear better solutions though.

Another is to wrap all values in a GenericRowData, but we'd need to iterate over the data to unwrap it.

yes, I was also thinking about storing null indexes

Hi @dawidwys, @snuyanzin,

I referenced LagAggFunction to use LinkedList to hold null values and it works. Now RESPECT NULLS and IGNORE NULLS are supported.

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/LagAggFunction.java

Lines 147 to 150 in 1ffb481

public static class LagAcc<T> {

public int offset = 1;

public T defaultValue = null;

public LinkedList<T> buffer = new LinkedList<>();

The downside of the solution is that the array must fit into memory at all times. The difference with the LagAggFunction is that LAG keeps at most n elements where n is controlled by the user.

Still I am reasonably good with the LinkedList approach because it anyhow needs to fit into memory when we emit it at the end as a single record. Writing this down for awareness.

snuyanzin · 2024-01-15T23:27:58Z

...e-planner/src/test/java/org/apache/flink/table/planner/functions/ArrayAggFunctionITCase.java

+                                Arrays.asList(
+                                        Row.of("A", new Integer[] {1, 2}),
+                                        Row.of("B", new Integer[] {2, 3}),
+                                        Row.of("C", new Integer[] {3}),


same here
based on input I would expect

Row.of("C", new Integer[] {3, null}),

Jiabao-Sun · 2024-01-16T06:27:53Z

Thanks @snuyanzin @dawidwys for the review.
Could you help review it again?

snuyanzin · 2024-01-16T11:45:06Z

...e-planner/src/test/java/org/apache/flink/table/planner/functions/ArrayAggFunctionITCase.java

+                TestSpec.forFunction(BuiltInFunctionDefinitions.ARRAY_AGG)
+                        .withDescription("ARRAY changelog stream aggregation")
+                        .withSource(
+                                ROW(STRING(), INT()),


One more thing
currently it checks only ability to work with INT input for ARRAY_AGG
it would be great to have tests for other types
especially ROW, ARRAY, MAP where expected output should be ARRAY<ROW>, ARRAY<ARRAY>, ARRAY<MAP>

Full types test is in ArrayAggFunctionTest.

snuyanzin · 2024-01-17T23:18:48Z

Thanks for addressing comments
in general it looks ok from my side
i guess there is one little thing: since it is based on Calcite parser it allows to have ORDER BY inside...
At the same time it is currently not supported on Flink level, not sure whether we can redefine this behavior however at least it would make sense to mention it in doc that it is not supported

Jiabao-Sun · 2024-01-18T02:21:05Z

Thanks for addressing comments in general it looks ok from my side i guess there is one little thing: since it is based on Calcite parser it allows to have ORDER BY inside... At the same time it is currently not supported on Flink level, not sure whether we can redefine this behavior however at least it would make sense to mention it in doc that it is not supported

Yes, ORDER BY allows sorting of any field in the input rows, but currently it is difficult to obtain the complete input rows for sorting in the function implementation. Therefore, the ORDER BY clause is not supported yet.
I have added an explanation in the documentation.

@snuyanzin, please help take a look again when you have time.

Jiabao-Sun · 2024-01-18T07:50:56Z

@flinkbot run azure

snuyanzin · 2024-01-18T11:24:50Z

docs/data/sql_functions.yml

+    description: |
+      By default or with keyword `ALL` and, return an array that concatenates the input rows
+      and returns `NULL` if there are no input rows. Use `DISTINCT` for one unique instance of each value.
+      By default null values are respected, use `IGNORE NULLS` to skip null values.


Suggested change

By default null values are respected, use `IGNORE NULLS` to skip null values.

By default `NULL` values are respected, use `IGNORE NULLS` to skip `NULL` values.

snuyanzin

Looks ok from my side

@dawidwys do you have anything to add (asking since you're also participating here)

Jiabao-Sun · 2024-01-18T14:33:46Z

@flinkbot run azure

Jiabao-Sun · 2024-01-22T09:31:23Z

Hi @dawidwys, please help review it again when you have time.
Thanks a lot.

dawidwys · 2024-02-02T14:34:30Z

Sorry, it takes such a long time from my side. I had a vacation in the meantime. I'll try to check it Monday. Nevertheless if you're comfortable with the PR @snuyanzin feel free to merge it without waiting for my review.

snuyanzin · 2024-02-02T14:59:31Z

I think we could wait until Monday or even more since right now there is a feature freeze and need to wait for cutting release branch

dawidwys

I put some comments to the implementation

dawidwys · 2024-02-05T09:49:54Z

...lanner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java


+    /**
+     * Use the definitions in Flink instead of {@link SqlLibraryOperators#ARRAY_AGG}, because we
+     * ignore nulls and returns nullable ARRAY type. Order by clause like <code>


is the comment still correct after the last changes?

dawidwys · 2024-02-05T10:28:36Z

...e-planner/src/test/java/org/apache/flink/table/planner/functions/ArrayAggFunctionITCase.java

+                                Arrays.asList(
+                                        Row.of("A", new Integer[] {1, 2}),
+                                        Row.of("B", new Integer[] {2, 2, 3}),
+                                        Row.of("C", new Integer[] {3}),


The downside of the solution is that the array must fit into memory at all times. The difference with the LagAggFunction is that LAG keeps at most n elements where n is controlled by the user.

Still I am reasonably good with the LinkedList approach because it anyhow needs to fit into memory when we emit it at the end as a single record. Writing this down for awareness.

dawidwys · 2024-02-05T10:38:12Z

...ntime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/ArrayAggFunction.java

+            List<T> retractBuffer = new ArrayList<>();
+            for (T element : acc.retractList) {
+                retractBuffer.add(element);
+            }
+            for (T element : otherAcc.retractList) {
+                retractBuffer.add(element);
+            }


do we need the retractBuffer? Can't we just iterate over both the retractList and create only the final newRetractBuffer? It seems we create quite some unnecessary objects and potentially list resizing here.

dawidwys · 2024-02-05T10:39:28Z

...ntime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/ArrayAggFunction.java

+
+            // update to acc
+            acc.list.clear();
+            acc.list.addAll(buffer);


can't we populate acc.list in a single go? The current approach does make sense with a ListView but it does not with a LinkedList kept in memory.

I believe we don't need the intermediate buffer and newRetractBuffer

dawidwys · 2024-02-05T10:41:06Z

...ntime/src/main/java/org/apache/flink/table/runtime/functions/aggregate/ArrayAggFunction.java

+    }
+
+    public void retract(ArrayAggAccumulator<T> acc, T value) throws Exception {
+        if (value != null) {


what about retracting nulls?

would be super nice to have a test case for that

dawidwys · 2024-02-06T16:36:34Z

Thanks for the update @Jiabao-Sun The implementation looks good now. I want to go through the tests again, but I need a bit more time. I hope this is fine, cause anyway we need to wait for a branch cut.

dawidwys

LGTM, thank you for the contribution @Jiabao-Sun !

Jiabao-Sun force-pushed the array_agg_function branch 5 times, most recently from ffb85dd to 8cee2f8 Compare September 16, 2023 17:34

deepyaman mentioned this pull request Nov 2, 2023

test(flink): revise WIP marks in aggregation tests ibis-project/ibis#7480

Merged

MartijnVisser requested a review from snuyanzin December 1, 2023 14:45

snuyanzin requested changes Dec 7, 2023

View reviewed changes

Jiabao-Sun force-pushed the array_agg_function branch 2 times, most recently from 72323f4 to 7fbdcf3 Compare December 18, 2023 08:24

Jiabao-Sun force-pushed the array_agg_function branch 3 times, most recently from 7991087 to a860e51 Compare December 19, 2023 02:38

Jiabao-Sun requested a review from snuyanzin December 19, 2023 09:55

Jiabao-Sun force-pushed the array_agg_function branch from a860e51 to 8ce0f58 Compare January 11, 2024 10:00

dawidwys self-assigned this Jan 11, 2024

dawidwys reviewed Jan 15, 2024

View reviewed changes

snuyanzin reviewed Jan 15, 2024

View reviewed changes

Jiabao-Sun force-pushed the array_agg_function branch from 8ce0f58 to 14222b4 Compare January 16, 2024 03:54

snuyanzin reviewed Jan 16, 2024

View reviewed changes

Jiabao-Sun force-pushed the array_agg_function branch from 207c452 to 68799f2 Compare January 18, 2024 02:16

Jiabao-Sun force-pushed the array_agg_function branch from 68799f2 to 7f6dd08 Compare January 18, 2024 08:06

snuyanzin reviewed Jan 18, 2024

View reviewed changes

snuyanzin approved these changes Jan 18, 2024

View reviewed changes

Jiabao-Sun requested a review from dawidwys January 19, 2024 06:12

dawidwys requested changes Feb 5, 2024

View reviewed changes

Jiabao-Sun added 8 commits February 5, 2024 20:08

[FLINK-21949][table] Support ARRAY_AGG aggregate function

c3d2839

Rebase master and resolve compilation problems

d46b2f3

Use equalityEvaluator to compare values

58bb064

Revert changes about equalityEvaluator

ccabc46

Support IGNORE NULLS and RESPECT NULLS

96c5c2c

Address Sergey's comments

a7e114f

Address comments

01051a2

Address Dawid's comments

d898519

Jiabao-Sun force-pushed the array_agg_function branch from 6e2dc00 to d898519 Compare February 6, 2024 08:06

dawidwys approved these changes Feb 7, 2024

View reviewed changes

dawidwys merged commit 042a4d2 into apache:master Feb 7, 2024

pnowojski pushed a commit to pnowojski/flink that referenced this pull request Mar 8, 2024

[FLINK-21949][table] Support ARRAY_AGG aggregate function (apache#23411)

ed87cc5

flinkbot added the component=TableSQL/API label Apr 4, 2024

	AggregateCall.create(
	call.getAggregation,
	false,
	false,
	false,
	call.getArgList,
	-1, // remove filterArg
	null,
	RelCollations.EMPTY,
	call.getType,
	call.getName)

	* <p>Note: Elements of a {@link ListView} must not be null. For heap-based state backends, {@code
	* hashCode/equals} of the original (i.e. external) class are used. However, the serialization
	* format will use internal data structures.

	public static class LagAcc<T> {
	public int offset = 1;
	public T defaultValue = null;
	public LinkedList<T> buffer = new LinkedList<>();

	By default null values are respected, use `IGNORE NULLS` to skip null values.
	By default `NULL` values are respected, use `IGNORE NULLS` to skip `NULL` values.

[FLINK-21949][table] Support ARRAY_AGG aggregate function #23411

[FLINK-21949][table] Support ARRAY_AGG aggregate function #23411

Uh oh!

Conversation

Jiabao-Sun commented Sep 13, 2023

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Sep 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Jiabao-Sun commented Oct 10, 2023

Uh oh!

snuyanzin left a comment

Choose a reason for hiding this comment

Uh oh!

Jiabao-Sun commented Dec 7, 2023

Uh oh!

Jiabao-Sun commented Dec 7, 2023

Uh oh!

snuyanzin commented Dec 7, 2023

Uh oh!

Jiabao-Sun commented Dec 18, 2023

Uh oh!

Jiabao-Sun commented Dec 19, 2023

Uh oh!

dawidwys commented Jan 15, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

snuyanzin commented Jan 15, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

snuyanzin Jan 16, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dawidwys Jan 16, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jiabao-Sun Jan 17, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jiabao-Sun commented Jan 16, 2024

Uh oh!

snuyanzin Jan 16, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

snuyanzin commented Jan 17, 2024

Uh oh!

Jiabao-Sun commented Jan 18, 2024

Uh oh!

Jiabao-Sun commented Jan 18, 2024

Uh oh!

flinkbot commented Sep 13, 2023 •

edited

Loading

snuyanzin Jan 16, 2024 •

edited

Loading

dawidwys Jan 16, 2024 •

edited

Loading

Jiabao-Sun Jan 17, 2024 •

edited

Loading

snuyanzin Jan 16, 2024 •

edited

Loading

snuyanzin Jan 18, 2024 •

edited

Loading

snuyanzin commented Feb 2, 2024 •

edited

Loading