[core] fix after compactioned the array elements are duplicated

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FullCompactionFileStoreITCase.java

@@ -28,19 +28,20 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
+import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** SQL ITCase for continuous file store. */
 public class FullCompactionFileStoreITCase extends CatalogITCaseBase {
     private final String table = "T";
+    private final String options =
+            " WITH('changelog-producer'='full-compaction', 'changelog-producer.compaction-interval' = '1s')";
 
     @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 IF NOT EXISTS T (a STRING, b STRING, c STRING, PRIMARY KEY (a) NOT ENFORCED)"
                         + options);
@@ -59,6 +60,28 @@ public void testStreamingRead() throws Exception {
         assertThat(iterator.collect(1)).containsExactlyInAnyOrder(Row.of("7", "8", "9"));
     }
 
+    /** Test streaming read with array and row nested data type. */
+    @Test
+    public void testStreamingReadOfArray() throws Exception {
+        String table = "T_ARRAY";
+        tEnv.executeSql(
+                "CREATE TABLE IF NOT EXISTS "
+                        + table
+                        + "("
+                        + "ID INT PRIMARY KEY NOT ENFORCED,\n"
+                        + "NAMES ARRAY<ROW<NAME STRING, MARK STRING>>\n"
+                        + ")"
+                        + options);
+        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);
+        assertThat(iterator.collect(1).stream().map(Row::toString).collect(Collectors.toList()))
+                .containsExactlyInAnyOrder("+I[1, [+I[c, mark1], +I[d, mark2], +I[e, mark3]]]");
+    }
+
     @Test
     public void testCompactedScanMode() throws Exception {
         BlockingIterator<Row, Row> iterator =

paimon-format/src/main/java/org/apache/paimon/format/orc/reader/OrcRowColumnVector.java

@@ -29,7 +29,7 @@
 public class OrcRowColumnVector extends AbstractOrcColumnVector
         implements org.apache.paimon.data.columnar.RowColumnVector {
 
-    private final ColumnarRow columnarRow;
+    private final VectorizedColumnBatch batch;
 
     public OrcRowColumnVector(StructColumnVector hiveVector, RowType type) {
         super(hiveVector);
@@ -38,12 +38,11 @@ public OrcRowColumnVector(StructColumnVector hiveVector, RowType type) {
         for (int i = 0; i < len; i++) {
             paimonVectors[i] = createPaimonVector(hiveVector.fields[i], type.getTypeAt(i));
         }
-        this.columnarRow = new ColumnarRow(new VectorizedColumnBatch(paimonVectors));
+        this.batch = new VectorizedColumnBatch(paimonVectors);
     }
 
     @Override
     public ColumnarRow getRow(int i) {
-        this.columnarRow.setRowId(i);
-        return this.columnarRow;
+        return new ColumnarRow(batch, i);
     }
 }