[CP] [CP] fix ps with timestamp parameter bug

src/observer/mysql/obmp_stmt_execute.cpp

@@ -1160,7 +1160,7 @@ int ObMPStmtExecute::parse_basic_param_value(ObIAllocator& allocator, const uint
     case MYSQL_TYPE_DATE:
     case MYSQL_TYPE_DATETIME:
     case MYSQL_TYPE_TIMESTAMP: {
-      if (OB_FAIL(parse_mysql_timestamp_value(static_cast<EMySQLFieldType>(type), data, param))) {
+      if (OB_FAIL(parse_mysql_timestamp_value(static_cast<EMySQLFieldType>(type), data, param, tz_info))) {
         LOG_WARN("parse timestamp value from client failed", K(ret));
       }
       break;
@@ -1453,7 +1453,8 @@ int ObMPStmtExecute::copy_or_convert_str(common::ObIAllocator& allocator, const
   return ret;
 }
 
-int ObMPStmtExecute::parse_mysql_timestamp_value(const EMySQLFieldType field_type, const char*& data, ObObj& param)
+int ObMPStmtExecute::parse_mysql_timestamp_value(
+    const EMySQLFieldType field_type, const char *&data, ObObj &param, const common::ObTimeZoneInfo *tz_info)
 {
   int ret = OB_SUCCESS;
   int8_t length = 0;
@@ -1524,7 +1525,12 @@ int ObMPStmtExecute::parse_mysql_timestamp_value(const EMySQLFieldType field_typ
   }
   if (OB_SUCC(ret)) {
     if (field_type == MYSQL_TYPE_TIMESTAMP) {
-      param.set_timestamp(value);
+      int64_t ts_value = 0;
+      if (OB_FAIL(ObTimeConverter::datetime_to_timestamp(value, tz_info, ts_value))) {
+        LOG_WARN("datetime to timestamp failed", K(ret));
+      } else {
+        param.set_timestamp(ts_value);
+      }
     } else if (field_type == MYSQL_TYPE_DATETIME) {
       param.set_datetime(value);
     } else if (field_type == MYSQL_TYPE_DATE) {

src/observer/mysql/obmp_stmt_execute.h

@@ -42,10 +42,11 @@ public:
 
   // Parse basic param value, no MYSQL_TYPE_COMPLEX or MYSQL_TYPE_CURSOR.
   // see parse_param_value()
-  static int parse_basic_param_value(ObIAllocator& allocator, const uint32_t type, const ObCharsetType charset,
-      const ObCollationType cs_type, const ObCollationType ncs_type, const char*& data,
-      const common::ObTimeZoneInfo* tz_info, ObObj& param);
-  static int parse_mysql_timestamp_value(const obmysql::EMySQLFieldType field_type, const char*& data, ObObj& param);
+  static int parse_basic_param_value(ObIAllocator &allocator, const uint32_t type, const ObCharsetType charset,
+      const ObCollationType cs_type, const ObCollationType ncs_type, const char *&data,
+      const common::ObTimeZoneInfo *tz_info, ObObj &param);
+  static int parse_mysql_timestamp_value(const obmysql::EMySQLFieldType field_type, const char *&data, ObObj &param,
+      const common::ObTimeZoneInfo *tz_info);
   static int parse_oracle_timestamp_value(
       const obmysql::EMySQLFieldType field_type, const char*& data, const ObTimeConvertCtx& cvrt_ctx, ObObj& param);
   static int parse_mysql_time_value(const char*& data, ObObj& param);

src/sql/code_generator/ob_static_engine_cg.cpp

@@ -1156,7 +1156,7 @@ int ObStaticEngineCG::fill_sort_funcs(
   if (OB_FAIL(sort_funcs.init(collations.count()))) {
     LOG_WARN("failed to init sort functions", K(ret));
   } else {
-    for (int64_t i = 0; i < collations.count(); ++i) {
+    for (int64_t i = 0; i < collations.count() && OB_SUCC(ret); ++i) {
       const ObSortFieldCollation& sort_collation = collations.at(i);
       ObExpr* expr = nullptr;
       if (OB_FAIL(sort_exprs.at(sort_collation.field_idx_, expr))) {

src/sql/engine/px/ob_px_coord.cpp

@@ -158,6 +158,7 @@ int ObPxCoord::rescan(ObExecContext& ctx) const
     LOG_WARN("get physical operator context failed", K(ret));
   } else if (OB_FAIL(terminate_running_dfos(ctx, px_ctx->coord_info_.dfo_mgr_))) {
     LOG_WARN("fail to release px resources in QC inner_close", K(ret));
+  } else if (FALSE_IT(clear_interrupt(px_ctx))) {
   } else if (OB_FAIL(destroy_all_channel(*px_ctx))) {
     LOG_WARN("release dtl channel failed", K(ret));
   } else if (FALSE_IT(unregister_first_buffer_cache(ctx, *px_ctx))) {
@@ -166,6 +167,8 @@ int ObPxCoord::rescan(ObExecContext& ctx) const
     LOG_WARN("failed to free allocator", K(ret));
   } else if (FALSE_IT(px_ctx->reset_for_rescan())) {
     // nop
+  } else if (OB_FAIL(register_interrupt(px_ctx))) {
+    LOG_WARN("fail to register interrupt", K(ret));
   } else if (OB_FAIL(init_dfo_mgr(ObDfoInterruptIdGen(px_ctx->interrupt_id_,
                                       (uint32_t)GCTX.server_id_,
                                       (uint32_t)px_ctx->qc_id_,

src/sql/engine/px/ob_px_coord_op.cpp

@@ -147,6 +147,7 @@ int ObPxCoordOp::rescan()
   ObDfo* root_dfo = NULL;
   if (OB_FAIL(terminate_running_dfos(coord_info_.dfo_mgr_))) {
     LOG_WARN("fail to release px resources in QC inner_close", K(ret));
+  } else if (FALSE_IT(clear_interrupt())) {
   } else if (OB_FAIL(destroy_all_channel())) {
     LOG_WARN("release dtl channel failed", K(ret));
   } else if (FALSE_IT(unregister_first_buffer_cache())) {
@@ -155,6 +156,8 @@ int ObPxCoordOp::rescan()
     LOG_WARN("failed to free allocator", K(ret));
   } else if (FALSE_IT(reset_for_rescan())) {
     // nop
+  } else if (OB_FAIL(register_interrupt())) {
+    LOG_WARN("fail to register interrupt", K(ret));
   } else if (OB_FAIL(init_dfo_mgr(ObDfoInterruptIdGen(interrupt_id_,
                                       (uint32_t)GCTX.server_id_,
                                       (uint32_t)(static_cast<const ObPxCoordSpec*>(&get_spec()))->qc_id_,