fix: Data insert not aligned with col names (risingwavelabs#6590)

* write test for bug

* found bug

* all permutations 3 columns

* all permutations 4 columns

* fix tests

* fix bug

* clean

* more cleanup

* remove inner

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

e2e_test/batch/basic/insert_expr.slt.part

@@ -110,7 +110,7 @@ select * from t;
 1 3 2
 1 2 3
 2 1 3
-2 3 1
+3 1 2
 
 statement ok
 insert into t (v3, v1, v2) values (1, 2, 3);
@@ -121,8 +121,8 @@ select * from t;
 1 3 2
 1 2 3
 2 1 3
-2 3 1
 3 1 2
+2 3 1
 
 statement ok
 insert into t (v3, v2, v1) values (1, 2, 3);
@@ -133,15 +133,126 @@ select * from t;
 1 3 2
 1 2 3
 2 1 3
-2 3 1
 3 1 2
+2 3 1
 3 2 1
 
 # column and table names have to be case insensitive
 
 statement ok
 drop table t;
 
+# Bug https://github.com/risingwavelabs/risingwave/issues/6542
+# Test all permutations
+
+statement ok
+create table t (v1 int, v2 int, v3 int, v4 int);
+
+statement ok
+insert into t (v1, v2, v3, v4) values (1, 2, 3, 4);
+
+statement ok
+insert into t (v1, v2, v4, v3) values (1, 2, 4, 3);
+
+statement ok
+insert into t (v1, v3, v2, v4) values (1, 3, 2, 4);
+
+statement ok
+insert into t (v1, v3, v4, v2) values (1, 3, 4, 2);
+
+statement ok
+insert into t (v1, v4, v2, v3) values (1, 4, 2, 3);
+
+statement ok
+insert into t (v1, v4, v3, v2) values (1, 4, 3, 2);
+
+statement ok
+insert into t (v2, v1, v3, v4) values (2, 1, 3, 4);
+
+statement ok
+insert into t (v2, v1, v4, v3) values (2, 1, 4, 3);
+
+statement ok
+insert into t (v2, v3, v1, v4) values (2, 3, 1, 4);
+
+statement ok
+insert into t (v2, v3, v4, v1) values (2, 3, 4, 1);
+
+statement ok
+insert into t (v2, v4, v1, v3) values (2, 4, 1, 3);
+
+statement ok
+insert into t (v2, v4, v3, v1) values (2, 4, 3, 1);
+
+statement ok
+insert into t (v3, v1, v2, v4) values (3, 1, 2, 4);
+
+statement ok
+insert into t (v3, v1, v4, v2) values (3, 1, 4, 2);
+
+statement ok
+insert into t (v3, v2, v1, v4) values (3, 2, 1, 4);
+
+statement ok
+insert into t (v3, v2, v4, v1) values (3, 2, 4, 1);
+
+statement ok
+insert into t (v3, v4, v1, v2) values (3, 4, 1, 2);
+
+statement ok
+insert into t (v3, v4, v2, v1) values (3, 4, 2, 1);
+
+statement ok
+insert into t (v4, v1, v2, v3) values (4, 1, 2, 3);
+
+statement ok
+insert into t (v4, v1, v3, v2) values (4, 1, 3, 2);
+
+statement ok
+insert into t (v4, v2, v1, v3) values (4, 2, 1, 3);
+
+statement ok
+insert into t (v4, v2, v3, v1) values (4, 2, 3, 1);
+
+statement ok
+insert into t (v4, v3, v1, v2) values (4, 3, 1, 2);
+
+statement ok
+insert into t (v4, v3, v2, v1) values (4, 3, 2, 1);
+
+query I rowsort
+select * from t;
+----
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+1 2 3 4
+
+statement ok
+drop table t;
+
+
+
 statement ok
 create table t (v1 int, v2 int, v3 int);
 

src/batch/src/executor/insert.rs

@@ -96,9 +96,9 @@ impl InsertExecutor {
 
             // No need to check for duplicate columns. This is already validated in binder
             if !&self.column_idxs.is_sorted() {
-                let mut ordered_cols: Vec<Column> = Vec::with_capacity(len);
-                for idx in &self.column_idxs {
-                    ordered_cols.push(columns[*idx].clone());
+                let mut ordered_cols: Vec<Column> = columns.clone();
+                for (i, idx) in self.column_idxs.iter().enumerate() {
+                    ordered_cols[*idx] = columns[i].clone()
                 }
                 columns = ordered_cols
             }