[Enhancement] replace table options to keep original dropped table in recycle bin

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -616,7 +616,7 @@ alterTableClause
         SET LEFT_PAREN partitionProperties=propertyItemList RIGHT_PAREN             #modifyPartitionClause
     | REPLACE partitions=partitionSpec? WITH tempPartitions=partitionSpec?
         FORCE? properties=propertyClause?                                           #replacePartitionClause
-    | REPLACE WITH TABLE name=identifier properties=propertyClause?                 #replaceTableClause
+    | REPLACE WITH TABLE name=identifier properties=propertyClause?  FORCE?         #replaceTableClause
     | RENAME newName=identifier                                                     #renameClause
     | RENAME ROLLUP name=identifier newName=identifier                              #renameRollupClause
     | RENAME PARTITION name=identifier newName=identifier                           #renamePartitionClause

fe/fe-core/src/main/cup/sql_parser.cup

@@ -1703,9 +1703,9 @@ alter_table_clause ::=
     {:
         RESULT = new ReplacePartitionClause(partitions, tempPartitions, isForce, properties);
     :}
-    | KW_REPLACE KW_WITH KW_TABLE ident:tblName opt_properties:properties
+    | KW_REPLACE KW_WITH KW_TABLE ident:tblName opt_properties:properties opt_force:force
     {:
-        RESULT = new ReplaceTableClause(tblName, properties);
+        RESULT = new ReplaceTableClause(tblName, properties, force);
     :}
     | KW_RENAME ident:newTableName
     {:

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -578,10 +578,12 @@ private void processReplaceTable(Database db, OlapTable origTable, List<AlterCla
         ReplaceTableClause clause = (ReplaceTableClause) alterClauses.get(0);
         String newTblName = clause.getTblName();
         boolean swapTable = clause.isSwapTable();
-        processReplaceTable(db, origTable, newTblName, swapTable);
+        boolean isForce = clause.isForce();
+        processReplaceTable(db, origTable, newTblName, swapTable, isForce);
     }
 
-    public void processReplaceTable(Database db, OlapTable origTable, String newTblName, boolean swapTable)
+    public void processReplaceTable(Database db, OlapTable origTable, String newTblName,
+                                    boolean swapTable, boolean isForce)
             throws UserException {
         db.writeLockOrDdlException();
         try {
@@ -598,10 +600,10 @@ public void processReplaceTable(Database db, OlapTable origTable, String newTblN
                 if (swapTable) {
                     origTable.checkAndSetName(newTblName, true);
                 }
-                replaceTableInternal(db, origTable, olapNewTbl, swapTable, false);
+                replaceTableInternal(db, origTable, olapNewTbl, swapTable, false, isForce);
                 // write edit log
                 ReplaceTableOperationLog log = new ReplaceTableOperationLog(db.getId(),
-                        origTable.getId(), olapNewTbl.getId(), swapTable);
+                        origTable.getId(), olapNewTbl.getId(), swapTable, isForce);
                 Env.getCurrentEnv().getEditLog().logReplaceTable(log);
                 LOG.info("finish replacing table {} with table {}, is swap: {}", oldTblName, newTblName, swapTable);
             } finally {
@@ -625,7 +627,7 @@ public void replayReplaceTable(ReplaceTableOperationLog log) throws MetaNotFound
         tableList.sort((Comparator.comparing(Table::getId)));
         MetaLockUtils.writeLockTablesOrMetaException(tableList);
         try {
-            replaceTableInternal(db, origTable, newTbl, log.isSwapTable(), true);
+            replaceTableInternal(db, origTable, newTbl, log.isSwapTable(), true, log.isForce());
         } catch (DdlException e) {
             LOG.warn("should not happen", e);
         } finally {
@@ -649,11 +651,10 @@ public void replayReplaceTable(ReplaceTableOperationLog log) throws MetaNotFound
      * 1.2 rename B to A, drop old A, and add new A to database.
      */
     private void replaceTableInternal(Database db, OlapTable origTable, OlapTable newTbl, boolean swapTable,
-                                      boolean isReplay)
+                                      boolean isReplay, boolean isForce)
             throws DdlException {
         String oldTblName = origTable.getName();
         String newTblName = newTbl.getName();
-
         // drop origin table and new table
         db.unregisterTable(oldTblName);
         db.unregisterTable(newTblName);
@@ -667,8 +668,15 @@ private void replaceTableInternal(Database db, OlapTable origTable, OlapTable ne
             origTable.checkAndSetName(newTblName, false);
             db.registerTable(origTable);
         } else {
+
             // not swap, the origin table is not used anymore, need to drop all its tablets.
-            Env.getCurrentEnv().onEraseOlapTable(origTable, isReplay);
+            // put original table to recycle bin.
+            if (isForce) {
+                Env.getCurrentEnv().onEraseOlapTable(origTable, isReplay);
+            } else {
+                Env.getCurrentRecycleBin().recycleTable(db.getId(), origTable, isReplay, isForce, 0);
+            }
+
             if (origTable.getType() == TableType.MATERIALIZED_VIEW) {
                 Env.getCurrentEnv().getMtmvService().deregisterMTMV((MTMV) origTable);
             }

fe/fe-core/src/main/java/org/apache/doris/analysis/ReplaceTableClause.java

@@ -37,11 +37,14 @@ public class ReplaceTableClause extends AlterTableClause {
     // if true, the new table and the old table will be exchanged.
     // default is true.
     private boolean swapTable;
+    // isForce used when swap is false, then original table need to keep in recycle bin or not.
+    private boolean isForce;
 
-    public ReplaceTableClause(String tblName, Map<String, String> properties) {
+    public ReplaceTableClause(String tblName, Map<String, String> properties, boolean isForce) {
         super(AlterOpType.REPLACE_TABLE);
         this.tblName = tblName;
         this.properties = properties;
+        this.isForce = isForce;
     }
 
     public String getTblName() {
@@ -52,6 +55,10 @@ public boolean isSwapTable() {
         return swapTable;
     }
 
+    public boolean isForce() {
+        return isForce;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) throws AnalysisException {
         if (Strings.isNullOrEmpty(tblName)) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/AlterMTMVReplaceInfo.java

@@ -89,7 +89,7 @@ public void run() throws UserException {
         Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(mvName.getDb());
         MTMV mtmv = (MTMV) db.getTableOrDdlException(mvName.getTbl(), TableType.MATERIALIZED_VIEW);
         MTMV newMtmv = (MTMV) db.getTableOrDdlException(newName, TableType.MATERIALIZED_VIEW);
-        Env.getCurrentEnv().getAlterInstance().processReplaceTable(db, mtmv, newName, swapTable);
+        Env.getCurrentEnv().getAlterInstance().processReplaceTable(db, mtmv, newName, swapTable, true);
         Env.getCurrentEnv().getMtmvService().alterTable(newMtmv, mvName.getTbl());
         if (swapTable) {
             Env.getCurrentEnv().getMtmvService().alterTable(mtmv, newName);

fe/fe-core/src/main/java/org/apache/doris/persist/ReplaceTableOperationLog.java

@@ -36,12 +36,15 @@ public class ReplaceTableOperationLog implements Writable {
     private long newTblId;
     @SerializedName(value = "swapTable")
     private boolean swapTable;
+    @SerializedName(value = "isForce")
+    private boolean isForce = true; // older version it was force. so keep same.
 
-    public ReplaceTableOperationLog(long dbId, long origTblId, long newTblId, boolean swapTable) {
+    public ReplaceTableOperationLog(long dbId, long origTblId, long newTblId, boolean swapTable, boolean isForce) {
         this.dbId = dbId;
         this.origTblId = origTblId;
         this.newTblId = newTblId;
         this.swapTable = swapTable;
+        this.isForce = isForce;
     }
 
     public long getDbId() {
@@ -60,6 +63,10 @@ public boolean isSwapTable() {
         return swapTable;
     }
 
+    public boolean isForce() {
+        return isForce;
+    }
+
     @Override
     public void write(DataOutput out) throws IOException {
         String json = GsonUtils.GSON.toJson(this);

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

@@ -3197,7 +3197,8 @@ private void handleOverwriteTable(InsertOverwriteTableStmt iotStmt) {
             List<AlterClause> ops = new ArrayList<>();
             Map<String, String> properties = new HashMap<>();
             properties.put("swap", "false");
-            ops.add(new ReplaceTableClause(tmpTableName.getTbl(), properties));
+            // swap false. but this operation is internal. so we will consider it as force drop for original table.
+            ops.add(new ReplaceTableClause(tmpTableName.getTbl(), properties, true));
             parsedStmt = new AlterTableStmt(targetTableName, ops);
             parsedStmt.setUserInfo(context.getCurrentUserIdentity());
             execute();

fe/fe-core/src/test/java/org/apache/doris/alter/AlterTest.java

@@ -995,7 +995,9 @@ public void testReplaceTable() throws Exception {
         Assert.assertEquals("replace2", replace2.getIndexNameById(replace2.getBaseIndexId()));
 
         // replace with no swap
-        replaceStmt = "ALTER TABLE test.replace1 REPLACE WITH TABLE replace2 properties('swap' = 'false')";
+        // tablet check will be done in this testcase. so
+        // we need to use force . behaviour same as older version
+        replaceStmt = "ALTER TABLE test.replace1 REPLACE WITH TABLE replace2 properties('swap' = 'false') force";
         alterTable(replaceStmt, false);
         replace1 = (OlapTable) db.getTableNullable("replace1");
         replace2 = (OlapTable) db.getTableNullable("replace2");

fe/fe-core/src/test/java/org/apache/doris/persist/ReplaceTableOperationLogTest.java

@@ -34,7 +34,7 @@ public void testSerialization() throws Exception {
         file.createNewFile();
         DataOutputStream dos = new DataOutputStream(new FileOutputStream(file));
 
-        ReplaceTableOperationLog log = new ReplaceTableOperationLog(1, 2, 3, true);
+        ReplaceTableOperationLog log = new ReplaceTableOperationLog(1, 2, 3, true, true);
         log.write(dos);
 
         dos.flush();

regression-test/data/catalog_recycle_bin_p0/test_replace_table_recycle.out

@@ -0,0 +1,21 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_check_1 --
+1	a	2022-01-02
+2	a	2023-01-02
+3	a	2024-01-02
+
+-- !select_check_2 --
+10	a	2022-01-02
+20	a	2023-01-02
+30	a	2024-01-02
+
+-- !select_check_3 --
+10	a	2022-01-02
+20	a	2023-01-02
+30	a	2024-01-02
+
+-- !select_check_4 --
+1	a	2022-01-02
+2	a	2023-01-02
+3	a	2024-01-02
+

regression-test/suites/catalog_recycle_bin_p0/test_replace_table_recycle.groovy

@@ -0,0 +1,91 @@
+// 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("test_replace_table_recycle") {
+    def table = "test_replace_table_recycle_t1"
+    def table2 = "test_replace_table_recycle_t2"
+    // create table and insert data
+    sql """ drop table if exists ${table} force"""
+    sql """ drop table if exists ${table2} force"""
+
+    sql """
+    create table ${table} (
+        `id` int(11),
+        `name` varchar(128),
+        `da` date
+    )
+    engine=olap
+    duplicate key(id)
+    partition by range(da)(
+        PARTITION p3 VALUES LESS THAN ('2023-01-01'),
+        PARTITION p4 VALUES LESS THAN ('2024-01-01'),
+        PARTITION p5 VALUES LESS THAN ('2025-01-01')
+    )
+    distributed by hash(id) buckets 2
+    properties(
+        "replication_num"="1",
+        "light_schema_change"="true"
+    );
+    """
+    sql """
+    create table ${table2} (
+        `id` int(11),
+        `name` varchar(128),
+        `da` date
+    )
+    engine=olap
+    duplicate key(id)
+    partition by range(da)(
+        PARTITION p3 VALUES LESS THAN ('2023-01-01'),
+        PARTITION p4 VALUES LESS THAN ('2024-01-01'),
+        PARTITION p5 VALUES LESS THAN ('2025-01-01')
+    )
+    distributed by hash(id) buckets 2
+    properties(
+        "replication_num"="1",
+        "light_schema_change"="true"
+    );
+    """    
+
+    sql """ insert into ${table} values(1, 'a', '2022-01-02'); """
+    sql """ insert into ${table} values(2, 'a', '2023-01-02'); """
+    sql """ insert into ${table} values(3, 'a', '2024-01-02'); """
+    sql """ SYNC;"""
+
+    sql """ insert into ${table2} values(10, 'a', '2022-01-02'); """
+    sql """ insert into ${table2} values(20, 'a', '2023-01-02'); """
+    sql """ insert into ${table2} values(30, 'a', '2024-01-02'); """
+    sql """ SYNC;"""
+
+    qt_select_check_1 """ select * from  ${table} order by id,name,da; """
+    qt_select_check_2 """ select * from  ${table2} order by id,name,da; """
+
+    sql """ ALTER TABLE ${table} REPLACE WITH table ${table2} PROPERTIES('swap' = 'false') ;"""
+
+    sql """ recover table ${table}  as ${table2}; """
+
+    qt_select_check_3 """ select * from  ${table} order by id,name,da; """
+    qt_select_check_4 """ select * from  ${table2} order by id,name,da; """
+
+    sql """ ALTER TABLE ${table} REPLACE WITH table ${table2} PROPERTIES('swap' = 'false') force;"""
+    // after force should not be able to recover.
+    assertThrows(Exception.class, {
+        sql """
+            recover table ${table}  as ${table2};
+            """
+    })    
+}