[Fix](Txn) Fix wrong columns sequence in txn model

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BatchInsertIntoTableCommand.java

@@ -30,6 +30,8 @@
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.trees.TreeNode;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.plans.Explainable;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.PlanType;
@@ -51,7 +53,10 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -143,15 +148,18 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
             Optional<PhysicalUnion> union = planner.getPhysicalPlan()
                     .<PhysicalUnion>collect(PhysicalUnion.class::isInstance).stream().findAny();
             if (union.isPresent()) {
-                InsertUtils.executeBatchInsertTransaction(ctx, targetTable.getQualifiedDbName(),
-                        targetTable.getName(), targetSchema, union.get().getConstantExprsList());
+                InsertUtils.executeBatchInsertTransaction(ctx, targetTable.getQualifiedDbName(), targetTable.getName(),
+                        targetSchema, reorderUnionData(sink.getOutputExprs(), union.get().getOutputs(),
+                                union.get().getConstantExprsList()));
                 return;
             }
             Optional<PhysicalOneRowRelation> oneRowRelation = planner.getPhysicalPlan()
                     .<PhysicalOneRowRelation>collect(PhysicalOneRowRelation.class::isInstance).stream().findAny();
             if (oneRowRelation.isPresent()) {
                 InsertUtils.executeBatchInsertTransaction(ctx, targetTable.getQualifiedDbName(),
-                        targetTable.getName(), targetSchema, ImmutableList.of(oneRowRelation.get().getProjects()));
+                        targetTable.getName(), targetSchema,
+                        ImmutableList.of(
+                                reorderOneRowData(sink.getOutputExprs(), oneRowRelation.get().getProjects())));
                 return;
             }
             // TODO: update error msg
@@ -161,6 +169,43 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
     }
 
+    // If table schema is c1, c2, c3, we do insert into table (c3, c2, c1) values(v3, v2, v1).
+    // The oneRowExprts are [v3#c1, v2#c2, v1#c3], which is wrong sequence. The sinkExprs are
+    // [v1#c3, v2#c2, v3#c1]. However, sinkExprs are SlotRefrence rather than Alias. We need to
+    // extract right sequence alias from oneRowExprs.
+    private List<NamedExpression> reorderOneRowData(List<NamedExpression> sinkExprs,
+            List<NamedExpression> oneRowExprs) {
+        List<NamedExpression> sequenceData = new ArrayList<>();
+        for (NamedExpression expr : sinkExprs) {
+            for (NamedExpression project : oneRowExprs) {
+                if (expr.getExprId().equals(project.getExprId())) {
+                    sequenceData.add(project);
+                    break;
+                }
+            }
+        }
+        return sequenceData;
+    }
+
+    private List<List<NamedExpression>> reorderUnionData(List<NamedExpression> sinkExprs,
+            List<NamedExpression> unionOutputs, List<List<NamedExpression>> unionExprs) {
+        Map<ExprId, Integer> indexMap = new HashMap<>();
+        for (int i = 0; i < unionOutputs.size(); i++) {
+            indexMap.put(unionOutputs.get(i).getExprId(), i);
+        }
+
+        List<List<NamedExpression>> reorderedExprs = new ArrayList<>();
+        for (List<NamedExpression> exprList : unionExprs) {
+            List<NamedExpression> reorderedList = new ArrayList<>();
+            for (NamedExpression expr : sinkExprs) {
+                int index = indexMap.get(expr.getExprId());
+                reorderedList.add(exprList.get(index));
+            }
+            reorderedExprs.add(reorderedList);
+        }
+        return reorderedExprs;
+    }
+
     @Override
     public StmtType stmtType() {
         return StmtType.INSERT;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalSink.java

@@ -54,4 +54,8 @@ public List<Slot> computeOutput() {
                 .map(NamedExpression::toSlot)
                 .collect(ImmutableList.toImmutableList());
     }
+
+    public List<NamedExpression> getOutputExprs() {
+        return outputExprs;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -2187,6 +2187,11 @@ private int executeForTxn(InsertStmt insertStmt)
         if (!(queryStmt instanceof SelectStmt)) {
             throw new TException("queryStmt is not SelectStmt, insert command error");
         }
+        if (((NativeInsertStmt) insertStmt).getTargetColumnNames() != null) {
+            throw new TException(
+                    "The legacy planner does not support specifying column names when using ·insert into values`."
+                            + " If you want to specify column names, please `set enable_nereids_planner=true`.");
+        }
         TransactionEntry txnEntry = context.getTxnEntry();
         SelectStmt selectStmt = (SelectStmt) queryStmt;
         int effectRows = 0;

regression-test/data/insert_p0/txn_insert_with_specify_columns.out

@@ -0,0 +1,21 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_unique --
+1	2	3	0
+11	\N	31	0
+12	22	1	0
+13	23	33	0
+14	\N	1	0
+15	25	35	1
+111	222	333	0
+1111	2222	3333	0
+
+-- !select_unique --
+1	2	3
+11	\N	31
+12	22	1
+13	23	33
+14	\N	1
+15	25	35
+111	222	333
+1111	2222	3333
+

regression-test/suites/data_model_p0/unique/test_unique_table_new_sequence.groovy

@@ -171,6 +171,8 @@ suite("test_unique_table_new_sequence") {
 
         qt_1 "select * from ${tableName} order by k1;"
 
+        sql """set enable_nereids_dml = true; """
+        sql """set enable_nereids_planner=true"""
         sql "begin;"
         sql "insert into ${tableName} (k1, v1, v2, v3, `OR`) values (2,20,20,20,20);"
         sql "commit;"

regression-test/suites/data_model_p0/unique/test_unique_table_sequence.groovy

@@ -173,6 +173,9 @@ suite("test_unique_table_sequence") {
         }
         sql "commit;"
 
+        sql """set enable_nereids_dml = true; """
+        sql """set enable_nereids_planner=true"""
+
         sql "begin;"
         sql "insert into ${tableName} (k1, v1, v2, v3, `or`, __doris_sequence_col__) values (1,1,1,1,1,1),(2,2,2,2,2,2),(3,3,3,3,3,3);"
         sql "commit;"

regression-test/suites/insert_p0/insert_group_commit_into.groovy

@@ -242,6 +242,9 @@ suite("insert_group_commit_into") {
                 assertEquals(23, rowCount[0][0])
 
                 // txn insert
+                sql """ set enable_nereids_dml = true; """
+                sql """ set enable_nereids_planner=true; """
+                sql """ set enable_fallback_to_original_planner=false; """
                 def stmt = prepareStatement """ begin  """
                 stmt.executeUpdate()
                 txn_insert """ insert into ${table}(id, name, score) values(20, 'i', 101);  """, 1

regression-test/suites/insert_p0/txn_insert_with_specify_columns.groovy

@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("txn_insert_with_specify_columns", "p0") {
+    def table = "txn_insert_with_specify_columns"
+
+    sql """ DROP TABLE IF EXISTS ${table}"""
+    sql """
+        CREATE TABLE ${table}(
+            c1 INT NULL,
+            c2 INT NULL,
+            c3 INT NULL default 1
+        ) ENGINE=OLAP
+        UNIQUE KEY(c1)
+        DISTRIBUTED BY HASH(c1) BUCKETS 3
+        PROPERTIES (
+            "replication_num" = "1"
+        );
+    """
+    sql """begin"""
+    sql """insert into ${table} (c1, c3, c2) values(1, 3, 2),(111,333,222),(1111,3333,2222);"""
+    sql """insert into ${table} (c3, c1) values(31, 11);"""
+    sql """insert into ${table} (c2, c1) values(22, 12);"""
+    sql """insert into ${table} (c2, c3, c1) values(23, 33, 13);"""
+    sql """insert into ${table} (c1) values(14);"""
+    sql """insert into ${table} (c3, c2, c1) values(35, 25, 15);"""
+    sql """insert into ${table} (__DORIS_DELETE_SIGN__, c3, c2, c1) values(1, 35, 25, 15);"""
+    sql """commit"""
+    sql """set show_hidden_columns=true"""
+    qt_select_unique """select c1,c2,c3,__DORIS_DELETE_SIGN__ from ${table} order by c1,c2,c3"""
+
+    sql """ DROP TABLE IF EXISTS ${table}"""
+    sql """
+        CREATE TABLE ${table}(
+            c1 INT NULL,
+            c2 INT NULL,
+            c3 INT NULL default 1
+        ) ENGINE=OLAP
+        DUPLICATE KEY(c1)
+        DISTRIBUTED BY HASH(c1) BUCKETS 3
+        PROPERTIES (
+            "replication_num" = "1"
+        );
+    """
+    sql """begin"""
+    sql """insert into ${table} (c1, c3, c2) values(1, 3, 2),(111,333,222),(1111,3333,2222);"""
+    sql """insert into ${table} (c3, c1) values(31, 11);"""
+    sql """insert into ${table} (c2, c1) values(22, 12);"""
+    sql """insert into ${table} (c2, c3, c1) values(23, 33, 13);"""
+    sql """insert into ${table} (c1) values(14);"""
+    sql """insert into ${table} (c3, c2, c1) values(35, 25, 15);"""
+    sql """commit"""
+    qt_select_unique """select c1,c2,c3 from ${table} order by c1,c2,c3"""
+
+    try {
+        sql """set enable_nereids_planner=false"""
+        sql """ DROP TABLE IF EXISTS ${table}"""
+        sql """
+            CREATE TABLE ${table}(
+                c1 INT NULL,
+                c2 INT NULL,
+                c3 INT NULL default 1
+            ) ENGINE=OLAP
+            UNIQUE KEY(c1)
+            DISTRIBUTED BY HASH(c1) BUCKETS 3
+            PROPERTIES (
+                "replication_num" = "1"
+            );
+        """
+        sql """begin"""
+        sql """insert into ${table} (c1, c3, c2) values(10, 30, 20);"""
+        logger.info(failed)
+        assertFalse(true);
+    } catch (Exception e) {
+        logger.info(e.getMessage())
+        assertTrue(e.getMessage().contains("The legacy planner does not support specifying column names"))
+    } finally {
+        sql "commit"
+        sql """ DROP TABLE IF EXISTS ${table}"""
+    }
+
+    try {
+        sql """set enable_nereids_planner=false"""
+        sql """ DROP TABLE IF EXISTS ${table}"""
+        sql """
+            CREATE TABLE ${table}(
+                c1 INT NULL,
+                c2 INT NULL,
+                c3 INT NULL default 1
+            ) ENGINE=OLAP
+            DUPLICATE KEY(c1)
+            DISTRIBUTED BY HASH(c1) BUCKETS 3
+            PROPERTIES (
+                "replication_num" = "1"
+            );
+        """
+        sql """begin"""
+        sql """insert into ${table} (c1, c3, c2) values(10, 30, 20);"""
+        logger.info(failed)
+        assertFalse(true);
+    } catch (Exception e) {
+        logger.info(e.getMessage())
+        assertTrue(e.getMessage().contains("The legacy planner does not support specifying column names"))
+    } finally {
+        sql "commit"
+        sql """ DROP TABLE IF EXISTS ${table}"""
+    }
+}