[Bug][Sort] Iceberg delete key cause ArrayIndexOutOfBoundsException

[thesumery]

What happened

Iceberg enable upsert mode, and delete key cause ArrayIndexOutOfBoundsException

What you expected to happen

Iceberg enable upsert mode, delete key success

How to reproduce

1. prepare a mysql 5.7+ and flink 1.13 and iceberg env
2. create a mysql table and insert some data

use inlong;
create table if not exists `inlong_input13` (
	`id` bigint,
	`name` varchar(63),
	PRIMARY KEY(id)
);
insert into inlong_input13 (id, name) values (1, 'haha');
insert into inlong_input13 (id, name) values (2, 'xixi');

3. create a iceberg table using hive

CREATE EXTERNAL TABLE inlong_iceberg13 (
    id bigint,
    name string
) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
  TBLPROPERTIES (
    "engine.hive.enabled"="true",
    'property-version'='2',
    'format-version'='2',
    'write.upsert.enabled'='true'
);

4. create a sort link in flink client

CREATE TABLE inlong_input13 (
     `id` bigint,
     `name` STRING,
     PRIMARY KEY(id) NOT ENFORCED
 ) WITH (
     'connector' = 'mysql-cdc-inlong',
     'hostname' = 'localhost',
     'port' = '3306',
     'username' = 'root',
     'password' = '123456',
     'database-name' = 'inlong',
     'table-name' = 'inlong_input13',
     'scan.incremental.snapshot.enabled'='true');
     
CREATE TABLE inlong_iceberg13  (
    id bigint,
    name string,
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector'='iceberg-inlong',
    'catalog-name'='hive_prod',
    'catalog-database'='inlong',
    'catalog-table'='inlong_iceberg13',
    'uri'='thrift://localhost:9083',
    'warehouse'='hdfs://localhost:8020/hive/warehouse'
);
insert into inlong_iceberg13 select * from inlong_input13;

Environment

No response

InLong version

master

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct