impl fixed route mode for temporary tables

src/share/system_variable/ob_sys_var_class_type.h

@@ -212,6 +212,7 @@ enum ObSysVarClassType {
   SYS_VAR_NLS_CURRENCY = 10111,
   SYS_VAR_NLS_ISO_CURRENCY = 10112,
   SYS_VAR_NLS_DUAL_CURRENCY = 10113,
+  SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED = 10116,
 };
 
 }

src/share/system_variable/ob_system_variable_alias.h

@@ -208,6 +208,7 @@ static const char* const OB_SV_PERFORMANCE_SCHEMA = "performance_schema";
 static const char* const OB_SV_NLS_CURRENCY = "nls_currency";
 static const char* const OB_SV_NLS_ISO_CURRENCY = "nls_iso_currency";
 static const char* const OB_SV_NLS_DUAL_CURRENCY = "nls_dual_currency";
+static const char* const OB_SV__OB_PROXY_SESSION_TEMPORARY_TABLE_USED = "_ob_proxy_session_temporary_table_used";
 
 }  // namespace share
 }  // namespace oceanbase

src/share/system_variable/ob_system_variable_factory.cpp

@@ -59,6 +59,7 @@ const char* ObSysVarFactory::SYS_VAR_NAMES_SORTED_BY_NAME[] = {"_enable_parallel
     "_force_parallel_query_dop",
     "_groupby_nopushdown_cut_ratio",
     "_nlj_batching_enabled",
+    "_ob_proxy_session_temporary_table_used",
     "_ob_px_bcast_optimization",
     "_ob_px_slave_mapping_threshold",
     "_ob_use_parallel_execution",
@@ -254,6 +255,7 @@ const ObSysVarClassType ObSysVarFactory::SYS_VAR_IDS_SORTED_BY_NAME[] = {SYS_VAR
     SYS_VAR__FORCE_PARALLEL_QUERY_DOP,
     SYS_VAR__GROUPBY_NOPUSHDOWN_CUT_RATIO,
     SYS_VAR__NLJ_BATCHING_ENABLED,
+    SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED,
     SYS_VAR__OB_PX_BCAST_OPTIMIZATION,
     SYS_VAR__OB_PX_SLAVE_MAPPING_THRESHOLD,
     SYS_VAR__OB_USE_PARALLEL_EXECUTION,
@@ -636,7 +638,8 @@ const char* ObSysVarFactory::SYS_VAR_NAMES_SORTED_BY_ID[] = {"auto_increment_inc
     "performance_schema",
     "nls_currency",
     "nls_iso_currency",
-    "nls_dual_currency"};
+    "nls_dual_currency",
+    "_ob_proxy_session_temporary_table_used"};
 
 bool ObSysVarFactory::sys_var_name_case_cmp(const char* name1, const ObString& name2)
 {
@@ -2958,6 +2961,17 @@ int ObSysVarFactory::create_sys_var(ObSysVarClassType sys_var_id, ObBasicSysVar*
         }
         break;
       }
+      case SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED: {
+        void* ptr = NULL;
+        if (OB_ISNULL(ptr = allocator_.alloc(sizeof(ObSysVarObProxySessionTemporaryTableUsed)))) {
+          ret = OB_ALLOCATE_MEMORY_FAILED;
+          LOG_ERROR("fail to alloc memory", K(ret), K(sizeof(ObSysVarObProxySessionTemporaryTableUsed)));
+        } else if (OB_ISNULL(sys_var_ptr = new (ptr) ObSysVarObProxySessionTemporaryTableUsed())) {
+          ret = OB_ALLOCATE_MEMORY_FAILED;
+          LOG_ERROR("fail to new ObSysVarObProxySessionTemporaryTableUsed", K(ret));
+        }
+        break;
+      }
 
       default: {
         ret = OB_ERR_UNEXPECTED;

src/share/system_variable/ob_system_variable_factory.h

@@ -2634,6 +2634,19 @@ public:
     return ObSysVariables::get_default_value(193);
   }
 };
+class ObSysVarObProxySessionTemporaryTableUsed : public ObBoolSysVar {
+public:
+  ObSysVarObProxySessionTemporaryTableUsed() : ObBoolSysVar(NULL, NULL, NULL, NULL, NULL)
+  {}
+  inline virtual ObSysVarClassType get_type() const
+  {
+    return SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED;
+  }
+  inline virtual const common::ObObj& get_global_default_value() const
+  {
+    return ObSysVariables::get_default_value(194);
+  }
+};
 
 class ObSysVarFactory {
 public:
@@ -2651,7 +2664,7 @@ public:
   static const common::ObString get_sys_var_name_by_id(ObSysVarClassType sys_var_id);
 
   const static int64_t MYSQL_SYS_VARS_COUNT = 81;
-  const static int64_t OB_SYS_VARS_COUNT = 113;
+  const static int64_t OB_SYS_VARS_COUNT = 114;
   const static int64_t ALL_SYS_VARS_COUNT = MYSQL_SYS_VARS_COUNT + OB_SYS_VARS_COUNT;
 
   const static int16_t OB_SPECIFIC_SYS_VAR_ID_OFFSET = 10000;

src/share/system_variable/ob_system_variable_init.cpp

@@ -2277,13 +2277,24 @@ static struct VarsInit {
     ObSysVarsIdToArrayIdx[SYS_VAR_NLS_DUAL_CURRENCY] = 193;
     ObSysVars[193].alias_ = "OB_SV_NLS_DUAL_CURRENCY";
 
+    ObSysVars[194].info_ = "this value is true if we have executed set transaction stmt, until a transaction "
+                           "commit(explicit or implicit) successfully";
+    ObSysVars[194].name_ = "_ob_proxy_session_temporary_table_used";
+    ObSysVars[194].data_type_ = ObIntType;
+    ObSysVars[194].value_ = "0";
+    ObSysVars[194].flags_ = ObSysVarFlag::READONLY | ObSysVarFlag::SESSION_SCOPE | ObSysVarFlag::INVISIBLE;
+    ObSysVars[194].id_ = SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED;
+    cur_max_var_id = MAX(cur_max_var_id, static_cast<int64_t>(SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED));
+    ObSysVarsIdToArrayIdx[SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED] = 194;
+    ObSysVars[194].alias_ = "OB_SV__OB_PROXY_SESSION_TEMPORARY_TABLE_USED";
+
     if (cur_max_var_id >= ObSysVarFactory::OB_MAX_SYS_VAR_ID) {
       HasInvalidSysVar = true;
     }
   }
 } vars_init;
 
-static int64_t var_amount = 194;
+static int64_t var_amount = 195;
 
 int64_t ObSysVariables::get_all_sys_var_count()
 {

src/share/system_variable/ob_system_variable_init.json

@@ -2582,5 +2582,17 @@
     "info_cn": "U数字格式元素的双货币符号",
     "background_cn": "兼容 oracle number format model U 数字格式元素",
     "ref_url": "https://yuque.antfin-inc.com/ob/product_functionality_review/efqk1r"
+  },
+  "_ob_proxy_session_temporary_table_used": {
+    "id": 10116,
+    "name": "_ob_proxy_session_temporary_table_used",
+    "value": "0",
+    "data_type": "bool",
+    "info": "this value is true if we have executed set transaction stmt, until a transaction commit(explicit or implicit) successfully",
+    "flags": "READONLY | SESSION | INVISIBLE",
+    "publish_version": "310",
+    "info_cn": "标记当前Session是否使用过Session级别临时表，用于告知proxy并修改路由决策",
+    "background_cn": "为了解决Session临时表断链接时可能导致的问题，当Session临时表第一次使用后，后续请求Proxy只会路由到同一个Session",
+    "ref_url": "https://yuque.antfin-inc.com/ob/product_functionality_review/hinq1e"
   }
 }

src/sql/code_generator/ob_code_generator_impl.cpp

@@ -375,6 +375,11 @@ int ObCodeGeneratorImpl::set_other_properties(const ObLogPlan& log_plan, ObPhysi
             if (table_schema->is_oracle_sess_tmp_table()) {
               phy_plan.set_contain_oracle_session_level_temporary_table();
             }
+            LOG_DEBUG("plan contain temporary table",
+                "trx level",
+                table_schema->is_oracle_trx_tmp_table(),
+                "session level",
+                table_schema->is_oracle_sess_tmp_table());
           }
         }
       }

src/sql/ob_sql.cpp

@@ -3093,6 +3093,19 @@ int ObSql::after_get_plan(ObPlanCacheCtx& pc_ctx, ObSQLSessionInfo& session, ObP
         }
       }
     }
+    if (NULL != phy_plan && !session.get_is_deserialized()) {
+      if (phy_plan->is_contain_oracle_session_level_temporary_table()) {
+        bool is_already_set = false;
+        if (OB_FAIL(session.get_session_temp_table_used(is_already_set))) {
+          LOG_WARN("fail to get session temp table used", K(ret));
+        } else if (is_already_set) {
+          // do nothing
+        } else if (OB_FAIL(session.set_session_temp_table_used(true))) {
+          LOG_WARN("fail to set session temp table used", K(ret));
+        }
+        LOG_DEBUG("plan contain oracle session level temporary table detected", K(is_already_set));
+      }
+    }
   } else {
     // not phy_plan, ignore
   }

src/sql/session/ob_basic_session_info.cpp

@@ -3584,7 +3584,8 @@ int ObBasicSessionInfo::is_sys_var_actully_changed(
       case SYS_VAR_OB_TRX_IDLE_TIMEOUT:
       case SYS_VAR_COLLATION_CONNECTION:
       case SYS_VAR_OB_PL_BLOCK_TIMEOUT:
-      case SYS_VAR_OB_COMPATIBILITY_MODE: {
+      case SYS_VAR_OB_COMPATIBILITY_MODE:
+      case SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED: {
         changed = old_val.get_meta() == new_val.get_meta() ? old_val != new_val : true;
       } break;
       default: {
@@ -3662,6 +3663,17 @@ int ObBasicSessionInfo::set_trans_specified(const bool is_spec)
   return ret;
 }
 
+int ObBasicSessionInfo::set_session_temp_table_used(const bool is_used)
+{
+  int ret = OB_SUCCESS;
+  ObObj obj;
+  obj.set_int(is_used);
+  if (OB_FAIL(update_sys_variable(SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED, obj))) {
+    LOG_WARN("fail to update_system_variable", K(ret));
+  }
+  return ret;
+}
+
 int ObBasicSessionInfo::set_safe_weak_read_snapshot_variable(const int64_t safe_snapshot)
 {
   int ret = OB_SUCCESS;
@@ -3836,6 +3848,10 @@ int ObBasicSessionInfo::get_sql_safe_updates(bool& v) const
   return get_bool_sys_var(SYS_VAR_SQL_SAFE_UPDATES, v);
 }
 
+int ObBasicSessionInfo::get_session_temp_table_used(bool& is_used) const
+{
+  return get_bool_sys_var(SYS_VAR__OB_PROXY_SESSION_TEMPORARY_TABLE_USED, is_used);
+}
 void ObBasicSessionInfo::reset_tx_variable()
 {
   // this function will be called in end_trans phase, and must be idempotent.

src/sql/session/ob_basic_session_info.h

@@ -1534,6 +1534,9 @@ public:
   }
   int load_default_sys_variable(int64_t var_idx);
 
+  int set_session_temp_table_used(const bool is_used);
+  int get_session_temp_table_used(bool& is_used) const;
+
 protected:
   int process_session_variable(
       share::ObSysVarClassType var, const common::ObObj& value, bool is_inc, const bool check_timezone_valid = true);