Check for rename collisions in finalize

Signed-off-by: mdouglas47 <[email protected]>

Author: morgando

schemachange/sc_rename_table.c

@@ -27,41 +27,24 @@
 int gbl_lightweight_rename = 0;
 int gbl_transactional_drop_plus_rename = 1;
 
-static int colliding_db_dropped_prior_in_my_tran(const struct schema_change_type * rename_sc)
-{
-    const struct schema_change_type * sc = rename_sc;
-    const char * const new_name = rename_sc->newtable;
-    while (sc) {
-        if (strcasecmp(sc->tablename, new_name) == 0 && sc->kind == SC_DROPTABLE) {
-            return 1;
-        }
-        sc = sc->sc_next;
-    }
-    return 0;
-}
-
-static int fatal_naming_collision(const struct dbtable *colliding_db, const struct dbtable *rename_db,
-                const struct schema_change_type * rename_sc)
-{
-    if (!colliding_db) { return 0; }
-    else if (gbl_lightweight_rename && (rename_db == colliding_db)) { return 0; }
-    else if (gbl_transactional_drop_plus_rename && (rename_sc->kind == SC_RENAMETABLE)
-        && colliding_db_dropped_prior_in_my_tran(rename_sc)) { return 0; }
-    return 1;
-}
-
 int do_rename_table(struct ireq *iq, struct schema_change_type *s,
                     tran_type *tran)
 {
-    struct dbtable *db;
+    struct dbtable *db, *colliding_db;
     iq->usedb = db = s->db = get_dbtable_by_name(s->tablename);
     if (db == NULL) {
         sc_client_error(s, "Table doesn't exist");
         return SC_TABLE_DOESNOT_EXIST;
     }
-
-    const struct dbtable *colliding_db = get_dbtable_by_name(s->newtable);
-    if (fatal_naming_collision(colliding_db, db, s)) {
+    /* If transactional_drop_plus_rename is enabled then we need to
+     * wait until finalize to check for colliding tables: If we check
+     * for colliding tables now, we incorrectly error on tables that were
+     * dropped prior in our transaction. If we check in finalize, those tables
+     * will have already been dropped.
+     */
+    if (!gbl_transactional_drop_plus_rename
+        && ((colliding_db = get_dbtable_by_name(s->newtable)) != NULL)
+        && (!gbl_lightweight_rename || (db != colliding_db))) {
         sc_client_error(s, "New table name exists");
         return SC_TABLE_ALREADY_EXIST;
     }
@@ -117,6 +100,12 @@ int finalize_rename_table(struct ireq *iq, struct schema_change_type *s,
         return -1;
     }
 
+    if (gbl_transactional_drop_plus_rename
+        && get_dbtable_by_name(s->newtable)) {
+        sc_client_error(s, "New table name exists");
+        return SC_TABLE_ALREADY_EXIST;
+    }
+
     char *newname = strdup(s->newtable);
     if (!newname) {
         sc_errf(s, "strdup error\n");

tests/renametable.test/reqoutput.txt

@@ -123,3 +123,7 @@
 [drop table t2] rc 0
 [alter table t2 rename to t] failed with rc 240 New table name exists
 [alter table t2 rename to t2] failed with rc 240 New table name exists
+[begin] rc 0
+[alter table t rename to dummy] rc 0
+[alter table t2 rename to t] rc 0
+[commit] rc 0

tests/renametable.test/reqoutput_lightweight.txt

@@ -123,3 +123,7 @@
 [drop table t2] rc 0
 [alter table t2 rename to t] failed with rc -3 New table name already exists
 [alter table t2 rename to t2] failed with rc -3 New table name already exists
+[commit] failed with rc -3 New table name already exists
+[begin] rc 0
+[alter table t rename to dummy] rc 0
+[alter table t2 rename to t] rc 0

tests/renametable.test/runit

@@ -133,6 +133,14 @@ cdb2sql ${CDB2_OPTIONS} $dbnm default "alter table t2 rename to t" &>> $OUT
 # Rename to our own name
 cdb2sql ${CDB2_OPTIONS} $dbnm default "alter table t2 rename to t2" &>> $OUT
 
+# Rename after renaming table with target name in txn
+cat <<EOF | cdb2sql ${CDB2_OPTIONS} $dbnm default - >> $OUT 2>&1
+begin
+alter table t rename to dummy\$\$
+alter table t2 rename to t\$\$
+commit
+EOF
+
 df=$(diff $OUT $EXP)
 if [ $? -ne 0 ] ; then
      echo "  ^^^^^^^^^^^^"