[Bug] after compactioned the array elements are duplicated

[MrTaozui]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

0.5
0.6
master

Compute Engine

flink

Minimal reproduce step

here is my uint test code.

package org.apache.paimon.flink;

import org.apache.paimon.utils.BlockingIterator;

import org.apache.flink.types.Row;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;

/** SQL ITCase for continuous file store. */
public class FullCompactionFileStoreITCase2 extends CatalogITCaseBase {
    private final String table = "test_array3";

    @Override
    @BeforeEach
    public void before() throws IOException {
        super.before();
        String options =
                " WITH('changelog-producer'='full-compaction', 'changelog-producer.compaction-interval' = '1s')";
        tEnv.executeSql(
                "create table test_array3\n"
                        + "(\n"
                        + "id int primary key not enforced,\n"
                        + "names array<row<name string,mark string>>\n"
                        + ")"
                        + options);
    }

    @Test
    public void testStreamingRead() throws Exception {
        BlockingIterator<Row, Row> iterator =
                BlockingIterator.of(streamSqlIter("SELECT * FROM %s", table));

        sql("insert into %s  values(1, array[('c','mark1'), ('d','mark2'), ('e','mark3')]);", table);
        for (Row row : iterator.collect(1)) {
            System.out.println(row);
        }
    }
}

What doesn't meet your expectations?

I expect it print :
+I[1, [+I[c, mark1], +I[d, mark2], +I[e, mark3]]]
but
it print :
+I[1, [+I[e, mark3], +I[e, mark3], +I[e, mark3]]]

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!