[BEAM-11800] Support ARRAY_AGG fn for Zetasql dialect

[sonam-vend]

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang				---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[robinyqiu]

I think here we expect schema:

Schema schema = Schema.builder().addArrayField("array_field", FieldType.of(FieldType.INT64)).build();

meaning it is an "array of int64"

[robinyqiu]

Here you simply created an array type without specifying the element type. I guess that's why the error message says the "inferred type is ARRAY NOT NULL" (just array type, your can ignore the "NOT NULL" suffix), where it should be "BIGINT NOT NULL ARRAY NOT NULL" (array of bigint).

To create an array type with element type specified, you may want to use createTypeFactory().createArrayType(), like here: https://github.com/robinyqiu/beam/blob/cbe87445d4259b6b485bc010231dda1895022d83/sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/ZetaSqlCalciteTranslationUtils.java#L170

[ibzib]

But ARRAY_AGG should support all element types, not just INT64. https://github.com/google/zetasql/blob/master/docs/aggregate_functions.md#array_agg

[sonam-vend]

@ibzib exactly. I try to implement generic array like not sure if it is the correct way.

private static RelDataType relDataType =
new JavaTypeFactoryImpl(RelDataTypeSystem.DEFAULT)
.builder()
.add("col_tinyint", SqlTypeName.TINYINT)
.add("col_smallint", SqlTypeName.SMALLINT)
.add("col_integer", SqlTypeName.INTEGER)
.add("col_bigint", SqlTypeName.BIGINT)
.add("col_float", SqlTypeName.FLOAT)
.add("col_double", SqlTypeName.DOUBLE)
.add("col_decimal", SqlTypeName.DECIMAL)
.add("col_string_varchar", SqlTypeName.VARCHAR)
.add("col_time", SqlTypeName.TIME)
.add("col_date", SqlTypeName.DATE)
.add("col_timestamp_with_local_time_zone", SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE)
.add("col_timestamp", SqlTypeName.TIMESTAMP)
.add("col_boolean", SqlTypeName.BOOLEAN)
.build();

public static final SqlOperator ARR_AGG_ARR_FN =
createUdafOperator(
"array_agg",
x -> createTypeFactory().createArrayType(relDataType, -1),
new UdafImpl<>(new ArrayAgg.ArrayAggArray()));

@robinyqiu

[robinyqiu]

Yeah, ARRAY_AGG should be generic. I said the type should be "BIGINT NOT NULL ARRAY NOT NULL" because that's from the error message.

Sonam, I believe there is some way that you can get the element type from the input x. You don't need to define your new mapping.

[sonam-vend]

x -> createTypeFactory().createArrayType(x.getOperandType(0), -1) worked.

Thanks @robinyqiu @ibzib

[robinyqiu]

The Beam ARRAY type expect the data to be a Collection type, not Object[] type (i.e Beam ARRAY != Java array). That's why you get the error during cast. The fix should be simple: change the return type of this function to list (also the third class generic parameter).

[sonam-vend]

R: @robinyqiu

[robinyqiu]

LGTM