Fix duplicated system table name problem

src/share/schema/ob_schema_getter_guard.cpp

@@ -1530,6 +1530,7 @@ int ObSchemaGetterGuard::get_table_id(uint64_t tenant_id,
       }
     } else {
       if (TEMP_TABLE_TYPE == check_type
+          && !is_inner_table(simple_table->get_table_id())
           && false == simple_table->is_tmp_table()) {
         // temporary table is not finded.
         LOG_TRACE("request for temporary table but non-temporary table returned", K(session_id_), K(session_id), K(check_type));

src/share/schema/ob_schema_mgr.cpp

@@ -3408,35 +3408,31 @@ int ObSchemaMgr::get_table_schema(
   const ObSimpleTableSchemaV2 *&table_schema) const
 {
   int ret = OB_SUCCESS;
-  bool is_system_table = false;
   table_schema = NULL;
 
   if (!check_inner_stat()) {
     ret = OB_NOT_INIT;
-    LOG_WARN("not init", K(ret));
+    LOG_WARN("not init", KR(ret));
   } else if (OB_INVALID_ID == tenant_id
              || OB_INVALID_ID == database_id
              || table_name.empty()) {
     ret = OB_INVALID_ARGUMENT;
-    LOG_WARN("invalid argument", K(ret), K(tenant_id), K(database_id), K(table_name));
-  } else if (OB_FAIL(ObSysTableChecker::is_sys_table_name(tenant_id, database_id, table_name, is_system_table))) {
-    LOG_WARN("fail to check if table is system table", K(ret), K(tenant_id), K(database_id), K(table_name));
+    LOG_WARN("invalid argument", KR(ret), K(tenant_id), K(database_id), K(table_name));
   } else if (OB_INVALID_TENANT_ID != tenant_id_
              && OB_SYS_TENANT_ID != tenant_id_
              && tenant_id_ != tenant_id) {
     ret = OB_INVALID_ARGUMENT;
-    LOG_WARN("tenant_id not matched", K(ret), K(tenant_id), K_(tenant_id));
+    LOG_WARN("tenant_id not matched", KR(ret), K(tenant_id), K_(tenant_id));
   } else {
     ObSimpleTableSchemaV2 *tmp_schema = NULL;
     ObNameCaseMode mode = OB_NAME_CASE_INVALID;
-    if (OB_SYS_TENANT_ID == tenant_id_ || is_system_table) {
-      // Scenarios for special handling of user tenant system tables
+    if (is_sys_tenant(tenant_id)) {
       mode = OB_ORIGIN_AND_INSENSITIVE;
     } else if (OB_FAIL(get_tenant_name_case_mode(tenant_id, mode))) {
-      LOG_WARN("fail to get_tenant_name_case_mode", K(tenant_id), K(ret));
+      LOG_WARN("fail to get_tenant_name_case_mode", K(tenant_id), KR(ret));
     } else if (OB_NAME_CASE_INVALID == mode) {
       ret = OB_ERR_UNEXPECTED;
-      LOG_WARN("invalid case mode", K(ret), K(mode));
+      LOG_WARN("invalid case mode", KR(ret), K(mode));
     }
     if (OB_SUCC(ret)) {
       const ObTableSchemaHashWrapper table_name_wrapper(tenant_id, database_id, session_id, mode, table_name);
@@ -3444,24 +3440,48 @@ int ObSchemaMgr::get_table_schema(
       if (OB_SUCCESS == hash_ret) {
         if (OB_ISNULL(tmp_schema)) {
           ret = OB_ERR_UNEXPECTED;
-          LOG_WARN("NULL ptr", K(ret), K(tmp_schema));
+          LOG_WARN("NULL ptr", KR(ret), K(table_name_wrapper));
         } else {
           table_schema = tmp_schema;
         }
       } else if (OB_HASH_NOT_EXIST == hash_ret && 0 != session_id && OB_INVALID_ID != session_id) {
         // If session_id != 0, the search just now is based on the possible match of the temporary table.
         // If it is not found, then it will be searched according to session_id = 0, which is the normal table.
-        const ObTableSchemaHashWrapper table_name_wrapper2(tenant_id, database_id, 0, mode, table_name);
-        hash_ret = table_name_map_.get_refactored(table_name_wrapper2, tmp_schema);
+        const ObTableSchemaHashWrapper table_name_wrapper1(tenant_id, database_id, 0, mode, table_name);
+        hash_ret = table_name_map_.get_refactored(table_name_wrapper1, tmp_schema);
         if (OB_SUCCESS == hash_ret) {
           if (OB_ISNULL(tmp_schema)) {
             ret = OB_ERR_UNEXPECTED;
-            LOG_WARN("NULL ptr", K(ret), K(tmp_schema));
+            LOG_WARN("NULL ptr", KR(ret), K(table_name_wrapper1));
           } else {
             table_schema = tmp_schema;
           }
         }
       }
+      // restrict creating duplicate table with existed inner table
+      if (OB_SUCC(ret) && OB_ISNULL(table_schema)) {
+         bool is_system_table = false;
+         if (OB_FAIL(ObSysTableChecker::is_sys_table_name(tenant_id, database_id, table_name, is_system_table))) {
+           LOG_WARN("fail to check if table is system table", KR(ret), K(tenant_id), K(database_id), K(table_name));
+         } else if (is_system_table) {
+           // Inner table's ObTableSchemaHashWrapper is stored with OB_ORIGIN_AND_INSENSITIVE. Actually,
+           // 1. For inner table in mysql database, comparision is insensitive.
+           // 2. For inner table in oracle database, comparision is sensitive.
+           const ObTableSchemaHashWrapper table_name_wrapper2(tenant_id, database_id,
+                                                              0, OB_ORIGIN_AND_INSENSITIVE, table_name);
+           hash_ret = table_name_map_.get_refactored(table_name_wrapper2, tmp_schema);
+           if (OB_SUCCESS == hash_ret) {
+             if (OB_ISNULL(tmp_schema)) {
+               ret = OB_ERR_UNEXPECTED;
+               LOG_WARN("NULL ptr", KR(ret), K(table_name_wrapper2));
+             } else {
+               table_schema = tmp_schema;
+             }
+           }
+         } else {
+           // not system table
+         }
+      }
     }
   }
 

src/share/schema/ob_schema_struct.cpp

@@ -219,7 +219,7 @@ int ObSysTableChecker::init_sys_table_name_map()
           } else if (OB_FAIL(sys_table_name_map_.set_refactored(key, value))) {
             LOG_WARN("fail to set table name array", K(ret), K(key), K(table));
           } else {
-            LOG_INFO("set tenant space table name", K(key), K(table));
+            LOG_INFO("set tenant space table name", K(key), K(table), "strlen", table_name.length());
           }
         } else if (OB_SUCCESS == ret) {
           if (OB_ISNULL(value)) {
@@ -323,7 +323,7 @@ int ObSysTableChecker::check_sys_table_name(
         is_system_table = (value->at(i) == table);
       }
     }
-    LOG_TRACE("check sys table name", K(ret), K(key), K(table));
+    LOG_TRACE("check sys table name", K(ret), K(key), K(table), "strlen", table_name.length());
   }
   return ret;
 }