Hey, we are currently exploring Flink SQL for one of our data sync use cases and we are struggling with Casts from MULTISET to ARRAY for a nested property. Let me briefly explain the (simplified) setup: • Person and Address are Flink tables based on two Kafka source topics backed by Debezium, • PersonUpdated is a sink table writing to a Kafka topic

CREATE TABLE Person (
    Id int,
    Firstname string,
    Lastname string,
    DateOfBirth date
) WITH (
  'connector' = 'kafka',
  'topic' = 'cdc.FlinkResearch.dbo.Person',
  ...
  'format' = 'debezium-avro-confluent',
  'debezium-avro-confluent.url' = '...'
);

CREATE TABLE Address (
    Id int,
    Street string,
    ZipCode string,
    City string,
    Country string,
    PersonId int
) WITH (
  'connector' = 'kafka',
  'topic' = 'cdc.FlinkResearch.dbo.Address',
  ...
  'format' = 'debezium-avro-confluent',
  'debezium-avro-confluent.url' = '...'
);

CREATE TABLE PersonUpdated (
    Id int,
    Firstname string,
    Lastname string,
    DateOfBirth date,
    Addresses ARRAY<ROW(Street string, City string, ZipCode string, Country string)>,

    primary key (Id) not enforced
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'cdc.FlinkResearch.dbo.PersonUpdated',
  ...
  'value.format' = 'avro-confluent',
  'value.avro-confluent.url' = '...',
  'key.format' = 'avro-confluent',
  'key.avro-confluent.url' = '...'
);

The idea is to create PersonUpdated entries for every person including the person's addresses as a list. We have tried various ways of casts in our (insert) queries, but still no success:

select p.Id, p.Firstname, p.Lastname, p.DateOfBirth, cast((select collect(row(Street, City, ZipCode, Country)) from Address where PersonId = p.Id) as ARRAY<ROW<Street string, City string, ZipCode string, Country string>>) as Addresses from Person p;
[ERROR] Could not execute SQL statement. Reason:
org.apache.calcite.sql.validate.SqlValidatorException: Cast function cannot convert value of type RecordType(VARCHAR(2147483647) EXPR$0, VARCHAR(2147483647) EXPR$1, VARCHAR(2147483647) EXPR$2, VARCHAR(2147483647) EXPR$3) MULTISET to type RecordType(VARCHAR(2147483647) Street, VARCHAR(2147483647) City, VARCHAR(2147483647) ZipCode, VARCHAR(2147483647) Country) ARRAY

Even more specific casts couldn't solve the issue:

select p.Id, p.Firstname, p.Lastname, p.DateOfBirth, cast((select collect(cast(row(Street, City, ZipCode, Country) as ROW<Street string, City string, ZipCode string, Country string>)) from Address where PersonId = p.Id) as ARRAY<ROW<Street string, City string, ZipCode string, Country string>>) as Addresses from Person p;
[ERROR] Could not execute SQL statement. Reason:
org.apache.calcite.sql.validate.SqlValidatorException: Cast function cannot convert value of type RecordType(VARCHAR(2147483647) Street, VARCHAR(2147483647) City, VARCHAR(2147483647) ZipCode, VARCHAR(2147483647) Country) MULTISET to type RecordType(VARCHAR(2147483647) Street, VARCHAR(2147483647) City, VARCHAR(2147483647) ZipCode, VARCHAR(2147483647) Country) ARRAY

Is there a way to cast nested multisets to an array with plain Flink SQL? Or are there alternatives to the

collect

function that would avoid casting at all?

I think what you're looking for is https://issues.apache.org/jira/browse/FLINK-21949 ?

Yes, this is exactly what I'm looking for. Thanks for the pointer. Is there any way to circumvent the current 'limitation'? Just using MULTISET in the table doesn't help as it would be treated as a map in Avro and this would require a string as a key 😞

@Gerald as an interim solution, you may be able to bridge the gap with a UDF. Wrote the one below to cast a multiset of strings to an array

package myproj.flink.udf;

import java.util.Map;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.ScalarFunction;

public class MultisetToStringArray extends ScalarFunction {
  @FunctionHint(input = @DataTypeHint("MULTISET<STRING>"), output = @DataTypeHint("ARRAY<STRING>"))
  public String[] eval(Map<String, Integer> multiSet) {
    return multiSet.keySet().toArray(String[]::new);
  }
}

Test SQL:

CREATE TEMPORARY FUNCTION MultisetToStringArray AS 'myproj.flink.udf.MultisetToStringArray' LANGUAGE JAVA;

CREATE TABLE source (
  group_id_gen INT,
  group_id AS 'group' || CAST(group_id_gen AS STRING),
  col_a_string STRING,
  col_b_int INT,
  col_c_timestamp TIMESTAMP,
  col_d_row ROW(
    row_d_1 STRING,
    row_d_2 STRING,
    row_d_3 INT
  ),
  col_e_array_string ARRAY<STRING>,
  col_f_array_bigint ARRAY<BIGINT>,
  col_g_array_row ARRAY<ROW(
    array_row_g_1 STRING,
    array_row_g_2 STRING,
    array_row_g_3 INT    
  )>
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.group_id_gen.min' = '1',
  'fields.group_id_gen.max' = '10',
  'fields.col_a_string.length' = '3',
  'fields.col_d_row.row_d_1.length' = '4',
  'fields.col_b_int.min' = '1',
  'fields.col_b_int.max' = '10000'
);

SELECT
  group_id,
  MultisetToStringArray(COLLECT(DISTINCT col_a_string))
FROM source
GROUP BY group_id;

Ah interesting, I guess in my case I would need to deal with

Object

instead of

String

as I can't infer the actual type of the ROW, similar to the

col_g_array_row

in your example?