close dblink connection when execute query failed

deps/oblib/src/lib/mysqlclient/ob_isql_connection.h

@@ -77,7 +77,8 @@ public:
        is_init_remote_env_(false),
        dblink_id_(OB_INVALID_ID),
        dblink_driver_proto_(-1),
-       has_reverse_link_credentials_(false)
+       has_reverse_link_credentials_(false),
+       exec_succ_(true)
   {}
   virtual ~ObISQLConnection() {}
 
@@ -133,12 +134,15 @@ public:
   bool get_init_remote_env() const { return is_init_remote_env_; }
   void set_reverse_link_creadentials(bool flag) { has_reverse_link_credentials_ = flag; }
   bool get_reverse_link_creadentials() { return has_reverse_link_credentials_; }
+  void set_exec_succ(bool flag) { exec_succ_ = flag; }
+  bool get_exec_succ() { return exec_succ_; }
 protected:
   bool oracle_mode_;
   bool is_init_remote_env_; // for dblink, we have to init remote env with some sql
   uint64_t dblink_id_; // for dblink, record dblink_id of a connection used by dblink
   int64_t dblink_driver_proto_; //for dblink, record DblinkDriverProto of a connection used by dblink
   bool has_reverse_link_credentials_; // for dblink, mark if this link has credentials set
+  bool exec_succ_;
 };
 
 } // end namespace sqlclient

deps/oblib/src/lib/mysqlclient/ob_mysql_connection_pool.cpp

@@ -444,6 +444,7 @@ int ObMySQLConnectionPool::acquire(const uint64_t tenant_id, ObMySQLConnection *
       LOG_WARN("fail to get connection", K(ret), K(tenant_id));
     } else if (OB_FAIL(try_connect(connection))) {
       LOG_WARN("failed to try connection, will release connection", K(ret), K(tenant_id));
+      connection->set_exec_succ(false);
       const bool succ = false;
       if (OB_SUCCESS != release(connection, succ)) { // ignore ret
         LOG_WARN("failed to release connection, ignore ret");
@@ -798,6 +799,7 @@ int ObMySQLConnectionPool::acquire_dblink(uint64_t dblink_id, const dblink_param
     LOG_WARN("fail to acquire dblink", K(ret), K(dblink_id));
   } else if (OB_FAIL(try_connect_dblink(dblink_conn, sql_request_level))) {
     LOG_WARN("fail to try connect dblink", K(ret), K(dblink_id));
+    dblink_conn->set_exec_succ(false);
     int release_ret = release_dblink(dblink_conn, sessid);
     if (release_ret != OB_SUCCESS) {
       LOG_WARN("fail to release dblink conn", K(release_ret), K(dblink_id));
@@ -810,7 +812,8 @@ int ObMySQLConnectionPool::acquire_dblink(uint64_t dblink_id, const dblink_param
 int ObMySQLConnectionPool::release_dblink(ObISQLConnection *dblink_conn, uint32_t sessid)
 {
   int ret = OB_SUCCESS;
-  if (OB_FAIL(release(dynamic_cast<ObMySQLConnection *>(dblink_conn), true, sessid))) {
+  const bool succ = OB_NOT_NULL(dblink_conn) ? dblink_conn->get_exec_succ() : false;
+  if (OB_FAIL(release(dynamic_cast<ObMySQLConnection *>(dblink_conn), succ, sessid))) {
     LOG_WARN("fail to release dblink conn", K(ret));
   }
   return ret;

deps/oblib/src/lib/mysqlclient/ob_mysql_proxy.cpp

@@ -512,6 +512,9 @@ int ObDbLinkProxy::execute_init_sql(ObISQLConnection *dblink_conn, int link_type
       }
     }
   }
+  if (OB_FAIL(ret) && OB_NOT_NULL(dblink_conn)) {
+    dblink_conn->set_exec_succ(false);
+  }
   return ret;
 }
 
@@ -539,6 +542,7 @@ int ObDbLinkProxy::dblink_read(ObISQLConnection *dblink_conn, ReadResult &result
     LOG_WARN("null ptr", K(ret), KP(dblink_conn), KP(sql));
   } else if (OB_FAIL(dblink_conn->execute_read(OB_INVALID_TENANT_ID, sql, result))) {
     LOG_WARN("read from dblink failed", K(ret), K(dblink_conn), KCSTRING(sql));
+    dblink_conn->set_exec_succ(false);
   } else {
     LOG_DEBUG("succ to read from dblink", K(sql));
   }
@@ -553,6 +557,7 @@ int ObDbLinkProxy::dblink_write(ObISQLConnection *dblink_conn, int64_t &affected
     LOG_WARN("null ptr", K(ret), KP(dblink_conn), KP(sql));
   } else if (OB_FAIL(dblink_conn->execute_write(OB_INVALID_TENANT_ID, sql, affected_rows))) {
     LOG_WARN("write to dblink failed", K(ret), K(dblink_conn), K(sql));
+    dblink_conn->set_exec_succ(false);
   } else {
     LOG_DEBUG("succ to write by dblink", K(sql));
   }
@@ -566,6 +571,7 @@ int ObDbLinkProxy::rollback(ObISQLConnection *dblink_conn)
     LOG_WARN("dblink conn is NULL", K(ret));
   } else if (OB_FAIL(dblink_conn->rollback())) {
     LOG_WARN("read from dblink failed", K(ret));
+    dblink_conn->set_exec_succ(false);
   }
   return ret;
 }

deps/oblib/src/lib/mysqlclient/ob_mysql_statement.cpp

@@ -92,6 +92,9 @@ int ObMySQLStatement::execute_update(int64_t &affected_rows)
     ret = OB_ERR_UNEXPECTED;
     LOG_ERROR("invalid mysql stmt", K_(conn), KP_(stmt), KP_(sql_str), K(ret));
   } else {
+    if (OB_UNLIKELY(!conn_->get_exec_succ())) {
+      LOG_ERROR("conn already failed, should not execute query again!", K(conn_));
+    }
     int64_t begin = ObTimeUtility::current_monotonic_raw_time();
     if (0 != (tmp_ret = mysql_real_query(stmt_, sql_str_, STRLEN(sql_str_)))) {
       ret = -mysql_errno(stmt_);
@@ -120,6 +123,9 @@ ObMySQLResult *ObMySQLStatement::execute_query()
     LOG_ERROR("invalid mysql stmt", K_(conn), K_(stmt), KP_(sql_str), K(ret));
   } else {
     int64_t begin = ObTimeUtility::current_monotonic_raw_time();
+    if (OB_UNLIKELY(!conn_->get_exec_succ())) {
+      LOG_ERROR("conn already failed, should not execute query again!", K(conn_));
+    }
     if (0 != mysql_real_query(stmt_, sql_str_, STRLEN(sql_str_))) {
       ret = -mysql_errno(stmt_);
       const int ER_LOCK_WAIT_TIMEOUT = -1205;