enable batch_stmt_optimization query return to large query queue, disable retry for non-first multi-stmt item

src/observer/mysql/ob_query_retry_ctrl.cpp

@@ -46,6 +46,7 @@ void ObRetryPolicy::try_packet_retry(ObRetryParam &v) const
     // muti stmt，并且不是第一句，不能扔回队列重试，因为前面的无法回滚
     v.retry_type_ = RETRY_TYPE_LOCAL;
   } else if (!THIS_WORKER.can_retry()) {
+    // false == THIS_WORKER.can_retry() means throw back to queue disabled by SOME logic
     v.retry_type_ = RETRY_TYPE_LOCAL;
   } else {
     v.retry_type_ = RETRY_TYPE_PACKET;

src/observer/mysql/obmp_query.cpp

@@ -481,7 +481,11 @@ int ObMPQuery::try_batched_multi_stmt_optimization(sql::ObSQLSessionInfo &sessio
                                          async_resp_used,
                                          need_disconnect))) {
     int tmp_ret = ret;
+    if (THIS_WORKER.need_retry()) {
+      // fail optimize, is a large query, just go back to large query queue and retry
+    } else {
       ret = OB_SUCCESS;
+    }
     LOG_WARN("failed to process batch stmt, cover the error code and reset retry flag",
         K(tmp_ret), K(ret), K(THIS_WORKER.need_retry()));
   } else {

src/observer/mysql/obmp_stmt_execute.cpp

@@ -1538,7 +1538,9 @@ int ObMPStmtExecute::try_batch_multi_stmt_optimization(ObSQLSessionInfo &session
     LOG_WARN("fail to trans_form extend type params_store", K(ret), K(arraybinding_size_));
   } else if (OB_FAIL(do_process_single(session, array_binding_params, has_more_result, force_sync_resp, async_resp_used))) {
     // 调用do_single接口
-    if (OB_BATCHED_MULTI_STMT_ROLLBACK == ret) {
+    if (THIS_WORKER.need_retry()) {
+      // just go back to large query queue and retry
+    } else if (OB_BATCHED_MULTI_STMT_ROLLBACK == ret) {
       LOG_TRACE("batched multi_stmt needs rollback", K(ret));
       ret = OB_SUCCESS;
     } else {

src/sql/ob_sql.cpp

@@ -2369,6 +2369,7 @@ OB_INLINE int ObSql::handle_text_query(const ObString &stmt, ObSqlCtx &context,
 
   int tmp_ret = ret;
   if (!is_begin_commit_stmt
+      && 0 == context.multi_stmt_item_.get_seq_num() /* only first item of a multi stmt, or single stmt */
       && OB_FAIL(handle_large_query(tmp_ret,
                                     result,
                                     ectx.get_need_disconnect_for_update(),