fix "Unexpected old row update or delete a non exist index row" for can_read_by_sql_no

src/storage/memtable/mvcc/ob_mvcc_iterator.cpp

@@ -310,16 +310,6 @@ int ObMvccValueIterator::read_by_sql_no(
   if (can_read_by_sql_no) {
     if (query_flag.is_read_latest()) {
       version_iter_ = iter;
-      if (NULL != version_iter_) {
-        ObMemtableCtx *curr_mt_ctx = static_cast<ObMemtableCtx *>(const_cast<ObIMvccCtx *>(&ctx));
-        transaction::ObTransCtx *trans_ctx = curr_mt_ctx->get_trans_ctx();
-        if (NULL != trans_ctx) {
-          if (!trans_ctx->is_bounded_staleness_read() && curr_mt_ctx->is_for_replay()) {
-            TRANS_LOG(WARN, "strong consistent read follower", K(trans_ctx->get_trans_id()), K(ctx));
-            ret = OB_NOT_MASTER;
-          }
-        }
-      }
     } else {
       if (!is_locked) {
         value->latch_.lock();
@@ -342,6 +332,18 @@ int ObMvccValueIterator::read_by_sql_no(
         }
       }
     }
+    if (NULL != version_iter_) {
+      ObMemtableCtx *curr_mt_ctx = static_cast<ObMemtableCtx *>(const_cast<ObIMvccCtx *>(&ctx));
+      if (curr_mt_ctx->is_for_replay()) {
+        transaction::ObTransCtx *trans_ctx = curr_mt_ctx->get_trans_ctx();
+        if (NULL != trans_ctx) {
+          if (!trans_ctx->is_bounded_staleness_read() && !trans_ctx->is_stmt_readonly()) {
+            TRANS_LOG(WARN, "strong consistent read follower", K(trans_ctx->get_trans_id()), K(ctx));
+            ret = OB_NOT_MASTER;
+          }
+        }
+      }
+    }
   }
 
   if (is_locked) {

src/storage/transaction/ob_trans_ctx.cpp

@@ -266,6 +266,20 @@ void ObTransCtx::print_trace_log_()
   FORCE_PRINT_TRACE(tlog_, "[force print]");
 }
 
+void ObTransCtx::set_cur_stmt_type(const sql::stmt::StmtType stmt_type, const bool is_sfu)
+{
+  if (is_sfu) {
+    cur_stmt_type_ = ObStmtType::SFU;
+  } else if (sql::stmt::T_SELECT == stmt_type ||
+              sql::stmt::T_BUILD_INDEX_SSTABLE == stmt_type) {
+    cur_stmt_type_ = ObStmtType::READ;
+  } else if (sql::stmt::T_NONE == stmt_type ) {
+    cur_stmt_type_ = ObStmtType::UNKNOWN;
+  } else {
+    cur_stmt_type_ = ObStmtType::WRITE;
+  }
+}
+
 void ObTransCtx::reset()
 {
   // unknown
@@ -299,6 +313,7 @@ void ObTransCtx::reset()
   is_dup_table_trans_ = false;
   is_exiting_ = false;
   is_readonly_ = false;
+  cur_stmt_type_ = ObStmtType::UNKNOWN;
   for_replay_ = false;
   need_print_trace_log_ = false;
   is_bounded_staleness_read_ = false;

src/storage/transaction/ob_trans_ctx.h

@@ -188,6 +188,7 @@ public:
         is_dup_table_trans_(false),
         is_exiting_(false),
         is_readonly_(false),
+        cur_stmt_type_(ObStmtType::UNKNOWN),
         for_replay_(false),
         need_print_trace_log_(false),
         is_bounded_staleness_read_(false),
@@ -229,6 +230,11 @@ public:
   {
     return is_readonly_;
   }
+  void set_cur_stmt_type(const sql::stmt::StmtType stmt_type, const bool is_sfu);
+  bool is_stmt_readonly() const
+  {
+    return cur_stmt_type_ == ObStmtType::READ;
+  }
   void set_bounded_staleness_read(const bool is_bounded_staleness_read)
   {
     is_bounded_staleness_read_ = is_bounded_staleness_read;
@@ -553,6 +559,7 @@ protected:
   bool is_dup_table_trans_;
   bool is_exiting_;
   bool is_readonly_;
+  int64_t cur_stmt_type_;
   bool for_replay_;
   bool need_print_trace_log_;
   bool is_bounded_staleness_read_;

src/storage/transaction/ob_trans_define.h

@@ -765,6 +765,14 @@ public:
   {
     return ((sql::stmt::T_SELECT == stmt_type_ && !is_sfu_) || sql::stmt::T_BUILD_INDEX_SSTABLE == stmt_type_);
   }
+  bool is_sfu() const
+  {
+    return is_sfu_;
+  }
+  sql::stmt::StmtType get_stmt_type() const
+  {
+    return stmt_type_;
+  }
   const char* get_sql_id() const
   {
     return sql_id_.ptr();
@@ -2268,6 +2276,21 @@ private:
   {}
 };
 
+class ObStmtType
+{
+public:
+  static const int64_t UNKNOWN = -1;
+  static const int64_t READ = 0;
+  static const int64_t SFU = 1;
+  static const int64_t WRITE = 2;
+public:
+  static bool is_valid(const int64_t stmt_type)
+  { return READ == stmt_type || SFU == stmt_type || WRITE == stmt_type; }
+private:
+  ObStmtType() {}
+  ~ObStmtType() {}
+};
+
 class Ob2PCState {
 public:
   static const int64_t UNKNOWN = -1;

src/storage/transaction/ob_trans_part_ctx.cpp

@@ -747,6 +747,7 @@ int ObPartTransCtx::start_task(const ObTransDesc& trans_desc, const int64_t snap
     pg_ = ob_partition;
     if (stmt_info_.main_stmt_change(sql_no) ||
         (cluster_version_before_2271_() && stmt_info_.main_stmt_change_compat(sql_no))) {
+      set_cur_stmt_type(trans_desc.get_cur_stmt_desc().get_stmt_type(), trans_desc.get_cur_stmt_desc().is_sfu());
       if (OB_FAIL(mt_ctx_.sub_trans_begin(snapshot_version_, expired_time, false, trx_lock_timeout))) {
         TRANS_LOG(ERROR,
             "sub transaction begin should never fail",