Hi! I'm hoping to get some advice on an issue we're having: • We are migrating from Spark to Flink for our company's data product pipelines (CDC might remain in Spark) • The Sink is Delta Table (Parquet format) • We absolutely need parity with what is being produced in the Delta Table by Spark Unfortunately that last point seems to be impossible at present. So the particular field in the Parquet record currently looks like this:

{
  ...
  "our_field_name": {
    "our_key_1": "our_value_1",
    "our_key_2": "our_value_2",
    ...
  }
}

In our Flink (Kotlin) application, we are defining the row as follows:

RowType.RowField(
    "our_field_name",
    MapType(
        VarCharType(VarCharType.MAX_LENGTH),
        VarCharType(VarCharType.MAX_LENGTH)
    )
)

... and we are setting the field with:

rowData.setField(
    OUR_FIELD_POSITION,
    GenericMapData(
        our_field.entries.associate {
            StringData.fromString(it.key) to StringData.fromString(it.value)
        }
    )
)

When we inspect the Parquet records however, we see the following is produced:

{
    ...
    "meta_info": {
        "key_value": [
            {
                "key": "providerCode",
                "value": "declined-provider-code"
            }
        ]
    }
}

I had a look at the source code here in the link below, and the

key_val

,

key

, and

val

that is added above is hardcoded when converting a `MAP`: https://github.com/apache/flink/blob/ffecb68b11e2cd914f92699112b5adb62a48724e/flin[…]/apache/flink/formats/parquet/utils/ParquetSchemaConverter.java I'd like to ask for advice on this issue. E.g. has anyone else had the requirement that we have and has anyone looked into this? Would it be possible to submit a pull request to address this? Thank you 🙂