hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Fix timezone bug in mview scheduler jobs

Browse Source
This commit is contained in:
obdev
2024-02-08 02:41:52 +00:00
committed by ob-robot
parent 48ef8c68fc
commit 1400476e92
6 changed files with 130 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
src/sql/resolver/ddl/ob_create_mlog_resolver.cpp
View File

@ -619,91 +619,54 @@ int ObCreateMLogResolver::resolve_purge_node(
return ret;
}
int ObCreateMLogResolver::resolve_const_expr_and_calc_value(ParseNode &node, ObObj &obj)
{
int ret = OB_SUCCESS;
ObRawExpr *const_expr = nullptr;
ParamStore params_array;
if (OB_FAIL(ObResolverUtils::resolve_const_expr(params_, node, const_expr, nullptr))) {
LOG_WARN("failed to resolve const expr", KR(ret));
} else if (OB_FAIL(ObSQLUtils::calc_const_expr(session_info_, *const_expr, obj, *allocator_, params_array))) {
LOG_WARN("failed to calc const expr", KR(ret));
}
return ret;
}
int ObCreateMLogResolver::resolve_purge_start_next_node(
ParseNode *purge_start_node,
ParseNode *purge_next_node,
ObCreateMLogStmt &create_mlog_stmt)
{
int ret = OB_SUCCESS;
ObObj start_time_obj;
ObObj current_time_obj;
int64_t current_time = 0;
int64_t current_time = ObTimeUtility::current_time() / 1000000L * 1000000L; // ignore micro seconds
int64_t start_time = current_time;
ObCreateMLogArg &create_mlog_arg = create_mlog_stmt.get_create_mlog_arg();
const char *current_time_expr_str = lib::is_oracle_mode() ? "current_date" : "sysdate()";
ObString current_time_expr(current_time_expr_str);
ObArenaAllocator tmp_allocator("mlog_expr");
if (OB_FAIL(ObMViewSchedJobUtils::calc_date_expression_from_str(
*session_info_, tmp_allocator, MTL_ID(), current_time_expr, current_time_obj))) {
LOG_WARN("failed to calc date expression from str", KR(ret), K(current_time_expr));
} else {
current_time = current_time_obj.get_timestamp();
}
if (OB_SUCC(ret) &&
OB_NOT_NULL(purge_start_node)
if (OB_NOT_NULL(purge_start_node)
&& (T_MLOG_PURGE_START_TIME_EXPR == purge_start_node->type_)
&& (1 == purge_start_node->num_child_)
&& OB_NOT_NULL(purge_start_node->children_)
&& OB_NOT_NULL(purge_start_node->children_[0])) {
ParseNode *purge_start_time_expr_node = purge_start_node->children_[0];
if (OB_FAIL(resolve_const_expr_and_calc_value(*purge_start_time_expr_node, start_time_obj))) {
LOG_WARN("failed to resolve default value", KR(ret));
if (OB_FAIL(ObMViewSchedJobUtils::resolve_date_expr_to_timestamp(params_,
*session_info_, *(purge_start_node->children_[0]), *allocator_, start_time))) {
LOG_WARN("failed to resolve date expr to timestamp", KR(ret));
} else if (start_time < current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter start date must evaluate to a time in the future",
KR(ret), K(current_time), K(start_time));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "start date");
}
}
if (OB_SUCC(ret)) {
if (start_time_obj.is_null()) {
start_time_obj = current_time_obj;
} else {
int64_t start_time = start_time_obj.get_timestamp();
if (start_time < current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter start date must evaluate to a time in the future",
KR(ret), K(current_time), K(start_time), K(current_time_obj), K(start_time_obj));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "start date");
}
}
if (OB_FAIL(ret)) {
} else if (OB_FAIL(ObMViewSchedJobUtils::convert_session_date_to_utc(session_info_,
start_time_obj, create_mlog_arg.purge_options_.start_datetime_expr_))) {
LOG_WARN("failed to convert session date to utc", KR(ret), K(start_time_obj));
} else {
create_mlog_stmt.set_purge_mode(ObMLogPurgeMode::DEFERRED);
}
if (OB_SUCC(ret)) {
create_mlog_arg.purge_options_.start_datetime_expr_.set_timestamp(start_time);
create_mlog_stmt.set_purge_mode(ObMLogPurgeMode::DEFERRED);
}
if (OB_SUCC(ret) && OB_NOT_NULL(purge_next_node)) {
ObObj next_time_obj;
if (OB_FAIL(resolve_const_expr_and_calc_value(*purge_next_node, next_time_obj))) {
LOG_WARN("failed to resolve default value", KR(ret));
int64_t next_time = 0;
if (OB_FAIL(ObMViewSchedJobUtils::resolve_date_expr_to_timestamp(params_,
*session_info_, *purge_next_node, *allocator_, next_time))) {
LOG_WARN("failed to resolve date expr to timestamp", KR(ret));
} else if (next_time <= current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter next date must evaluate to a time in the future",
KR(ret), K(current_time), K(next_time));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "next date");
} else {
int64_t next_time = next_time_obj.get_timestamp();
if (next_time <= current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter next date must evaluate to a time in the future",
KR(ret), K(current_time), K(next_time), K(current_time_obj), K(next_time_obj));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "next date");
ObString next_date_str(purge_next_node->str_len_, purge_next_node->str_value_);
if (OB_FAIL(ob_write_string(*allocator_, next_date_str,
create_mlog_arg.purge_options_.next_datetime_expr_))) {
LOG_WARN("fail to write string", KR(ret));
}
}
if (OB_SUCC(ret)) {
create_mlog_arg.purge_options_.next_datetime_expr_.assign_ptr(
const_cast<char *>(purge_next_node->str_value_),
static_cast<int32_t>(purge_next_node->str_len_));
create_mlog_stmt.set_purge_mode(ObMLogPurgeMode::DEFERRED);
}
}
return ret;
}

1
src/sql/resolver/ddl/ob_create_mlog_resolver.h
View File

@ -55,7 +55,6 @@ protected:
private:
bool is_column_exists(ObIArray<ObString> &column_name_array, const ObString &column_name);
int resolve_const_expr_and_calc_value(ParseNode &node, ObObj &obj);
private:
enum ParameterEnum {

88
src/sql/resolver/ddl/ob_create_view_resolver.cpp
View File

@ -998,24 +998,6 @@ int ObCreateViewResolver::create_alias_names_auto(
return ret;
}
int ObCreateViewResolver::resolve_const_expr_and_calc_value(ParseNode &node, ObObj &obj)
{
int ret = OB_SUCCESS;
ObRawExpr *const_expr = nullptr;
ParamStore params_array;
if (OB_FAIL(ObResolverUtils::resolve_const_expr(params_, node, const_expr, nullptr))) {
LOG_WARN("fail to resolve const expr", KR(ret));
} else if (OB_FAIL(ObSQLUtils::calc_const_expr(session_info_, *const_expr, obj, *allocator_, params_array))) {
LOG_WARN("fail to calc const expr", KR(ret));
} else if (!obj.is_datetime()) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid datetime expression", KR(ret), K(obj));
LOG_USER_ERROR(OB_INVALID_ARGUMENT, "datetime expression");
}
return ret;
}
int ObCreateViewResolver::resolve_mv_refresh_info(ParseNode *refresh_info_node,
ObMVRefreshInfo &refresh_info)
{
@ -1094,64 +1076,40 @@ int ObCreateViewResolver::resolve_mv_refresh_info(ParseNode *refresh_info_node,
} else {
ParseNode *start_date = refresh_interval_node->children_[0];
ParseNode *next_date = refresh_interval_node->children_[1];
ObObj start_time_obj;
ObObj current_time_obj;
int64_t current_time = 0;
const char *current_time_expr_str = lib::is_oracle_mode() ? "current_date" : "sysdate()";
ObString current_time_expr(current_time_expr_str);
ObArenaAllocator tmp_allocator;
if (OB_FAIL(ObMViewSchedJobUtils::calc_date_expression_from_str(
*session_info_, tmp_allocator, MTL_ID(), current_time_expr, current_time_obj))) {
LOG_WARN("failed to calc date expression from str", KR(ret), K(current_time_expr));
} else {
current_time = current_time_obj.get_timestamp();
}
int64_t current_time = ObTimeUtility::current_time() / 1000000L * 1000000L; // ignore micro seconds
int64_t start_time = current_time;
if (OB_SUCC(ret)
&& OB_NOT_NULL(start_date)
if (OB_NOT_NULL(start_date)
&& (T_MV_REFRESH_START_EXPR == start_date->type_)
&& (1 == start_date->num_child_)
&& (OB_NOT_NULL(start_date->children_))
&& (OB_NOT_NULL(start_date->children_[0]))) {
if (OB_FAIL(resolve_const_expr_and_calc_value(*(start_date->children_[0]),
start_time_obj))) {
LOG_WARN("fail to resolve default value", KR(ret));
if (OB_FAIL(ObMViewSchedJobUtils::resolve_date_expr_to_timestamp(params_,
*session_info_, *(start_date->children_[0]), *allocator_, start_time))) {
LOG_WARN("failed to resolve date expr to timestamp", KR(ret));
} else if (start_time < current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter start date must evaluate to a time in the future",
KR(ret), K(current_time), K(start_time));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "start date");
}
}
if (OB_SUCC(ret)) {
if (start_time_obj.is_null()) {
start_time_obj = current_time_obj;
} else {
int64_t start_time = start_time_obj.get_timestamp();
if (start_time < current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter start date must evaluate to a time in the future",
KR(ret), K(current_time), K(start_time), K(current_time_obj), K(start_time_obj));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "start date");
}
}
if (OB_FAIL(ret)) {
} else if (OB_FAIL(ObMViewSchedJobUtils::convert_session_date_to_utc(
session_info_, start_time_obj, refresh_info.start_time_))) {
LOG_WARN("failed to convert session date to utc", KR(ret), K(start_time_obj));
}
refresh_info.start_time_.set_timestamp(start_time);
}
if (OB_SUCC(ret) && next_date != nullptr) {
ObObj next_time_obj;
if (OB_FAIL(resolve_const_expr_and_calc_value(*next_date, next_time_obj))) {
LOG_WARN("fail to resolve default value", KR(ret));
if (OB_SUCC(ret) && OB_NOT_NULL(next_date)) {
int64_t next_time = 0;
if (OB_FAIL(ObMViewSchedJobUtils::resolve_date_expr_to_timestamp(params_,
*session_info_, *next_date, *allocator_, next_time))) {
LOG_WARN("fail to resolve date expr to timestamp", KR(ret));
} else if (next_time <= current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter next date must evaluate to a time in the future",
KR(ret), K(current_time), K(next_time));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "next date");
} else {
int64_t next_time = next_time_obj.get_timestamp();
if (next_time <= current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter next date must evaluate to a time in the future",
KR(ret), K(current_time), K(next_time), K(current_time_obj), K(next_time_obj));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "next date");
}
}
if (OB_SUCC(ret)) {
ObString next_date_str(next_date->str_len_, next_date->str_value_);
if (OB_FAIL(ob_write_string(*allocator_, next_date_str, refresh_info.next_time_expr_))) {
LOG_WARN("fail to write string", KR(ret));

1
src/sql/resolver/ddl/ob_create_view_resolver.h
View File

@ -71,7 +71,6 @@ private:
ObSelectStmt *select_stmt);
int resolve_column_list(ParseNode *view_columns_node,
common::ObIArray<common::ObString> &column_list);
int resolve_const_expr_and_calc_value(ParseNode &node, ObObj &obj);
int resolve_mv_refresh_info(ParseNode *refresh_info_node,
ObMVRefreshInfo &refresh_info);

135
src/storage/mview/ob_mview_sched_job_utils.cpp
View File

@ -297,23 +297,22 @@ int ObMViewSchedJobUtils::disable_mlog_purge_job(
return ret;
}
int ObMViewSchedJobUtils::calc_date_expression_from_str(
int ObMViewSchedJobUtils::calc_date_expr_from_str(
sql::ObSQLSessionInfo &session,
ObIAllocator &allocator,
const uint64_t tenant_id,
const ObString &interval_str,
ObObj &date_obj)
int64_t &timestamp)
{
int ret = OB_SUCCESS;
ObSchemaGetterGuard schema_guard;
ObSchemaChecker schema_checker;
CK (OB_NOT_NULL(GCTX.schema_service_));
OZ (GCTX.schema_service_->get_tenant_schema_guard(tenant_id, schema_guard));
OZ (schema_checker.init(schema_guard));
SMART_VAR(ObResolverParams, resolver_ctx){
const ParseNode *expr_node = nullptr;
ObRawExpr *const_expr = nullptr;
ParamStore params_array;
ObRawExprFactory expr_factory(allocator);
ObSchemaChecker schema_checker;
ObSchemaGetterGuard schema_guard;
OZ (GCTX.schema_service_->get_tenant_schema_guard(tenant_id, schema_guard));
schema_checker.init(schema_guard);
resolver_ctx.allocator_ = &allocator;
resolver_ctx.schema_checker_ = &schema_checker;
resolver_ctx.session_info_ = &session;
@ -325,44 +324,26 @@ int ObMViewSchedJobUtils::calc_date_expression_from_str(
allocator,
expr_node))) {
LOG_WARN("failed to parse default expr from str", KR(ret), K(interval_str));
} else if (OB_FAIL(sql::ObResolverUtils::resolve_const_expr(resolver_ctx,
*expr_node,
const_expr,
nullptr))) {
LOG_WARN("fail to resolve const expr", KR(ret));
} else if (OB_FAIL(sql::ObSQLUtils::calc_const_expr(&session,
*const_expr, date_obj, allocator, params_array))) {
LOG_WARN("fail to calc const expr", KR(ret));
} else if (OB_ISNULL(expr_node)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("expr node is null", KR(ret));
} else if (OB_FAIL(resolve_date_expr_to_timestamp(resolver_ctx,
session,
*expr_node,
allocator,
timestamp))) {
LOG_WARN("failed to resolve data expr to timestamp", KR(ret));
} else {
LOG_INFO("dbms_sched_table_operator calc_date_expression_from_str end",
KR(ret), K(interval_str), K(date_obj));
LOG_INFO("mview_sched_job_utils calc_date_expr_from_str end",
KR(ret), K(tenant_id), K(interval_str), K(timestamp));
}
}
return ret;
}
int ObMViewSchedJobUtils::convert_session_date_to_utc(
ObSQLSessionInfo *session_info,
const ObObj &in_obj,
ObObj &out_obj)
{
int ret = OB_SUCCESS;
int64_t date_utc_ts = 0;
ObTime ob_time;
ObTimeConvertCtx time_cvrt_ctx(TZ_INFO(session_info), true);
OZ (ObTimeConverter::datetime_to_ob_time(in_obj.get_datetime(),
TZ_INFO(session_info), ob_time));
OZ (ObTimeConverter::ob_time_to_utc(ObTimestampTZType, time_cvrt_ctx, ob_time));
OZ (ObTimeConverter::ob_time_to_datetime(ob_time, time_cvrt_ctx, date_utc_ts));
if (OB_SUCC(ret)) {
out_obj.set_timestamp(date_utc_ts);
}
return ret;
}
int ObMViewSchedJobUtils::calc_date_expression(
ObDBMSSchedJobInfo &job_info,
int64_t &next_date_utc_ts)
int64_t &next_date_ts)
{
int ret = OB_SUCCESS;
const uint64_t tenant_id = job_info.get_tenant_id();
@ -384,38 +365,20 @@ int ObMViewSchedJobUtils::calc_date_expression(
THIS_WORKER.set_compatibility_mode(Worker::CompatMode::ORACLE);
}
ObObj next_time_obj;
ObObj current_time_obj;
const char *current_time_expr_str = is_oracle_tenant ? "current_date" : "sysdate()";
ObString current_time_expr(current_time_expr_str);
int64_t current_time = ObTimeUtility::current_time() / 1000000L * 1000000L; // ignore micro seconds
int64_t next_time = 0;
ObArenaAllocator tmp_allocator("MVSchedTmp");
if (OB_FAIL(calc_date_expression_from_str(session, tmp_allocator,
tenant_id, job_info.get_interval(), next_time_obj))) {
if (OB_FAIL(calc_date_expr_from_str(session, tmp_allocator,
tenant_id, job_info.get_interval(), next_time))) {
LOG_WARN("failed to calc date expression from str", KR(ret),
K(tenant_id), K(job_info.get_interval()));
} else if (OB_FAIL(calc_date_expression_from_str(session, tmp_allocator,
tenant_id, current_time_expr, current_time_obj))) {
LOG_WARN("failed to calc date expression from str", KR(ret),
K(tenant_id), K(current_time_expr));
} else if (next_time <= current_time) {
ret = OB_ERR_TIME_EARLIER_THAN_SYSDATE;
LOG_WARN("the parameter next date must evaluate to a time in the future",
KR(ret), K(current_time), K(next_time));
LOG_USER_ERROR(OB_ERR_TIME_EARLIER_THAN_SYSDATE, "next date");
} else {
int64_t next_time = next_time_obj.get_timestamp();
const int64_t current_time = current_time_obj.get_timestamp();
if (next_time <= current_time) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("next date must evaluated to a time in the future",
KR(ret), K(current_time), K(next_time), K(current_time_obj), K(next_time_obj));
} else {
int64_t next_date_ts = 0;
ObTime ob_time;
ObTimeConvertCtx time_cvrt_ctx(TZ_INFO(&session), true);
OZ (ObTimeConverter::datetime_to_ob_time(next_time_obj.get_datetime(),
session.get_timezone_info(), ob_time));
OZ (ObTimeConverter::ob_time_to_utc(ObTimestampTZType, time_cvrt_ctx, ob_time));
OZ (ObTimeConverter::ob_time_to_datetime(ob_time, time_cvrt_ctx, next_date_ts));
if (OB_SUCC(ret)) {
next_date_utc_ts = next_date_ts;
}
}
next_date_ts = next_time;
}
if (is_oracle_tenant && !is_oracle_mode) {
@ -427,5 +390,45 @@ int ObMViewSchedJobUtils::calc_date_expression(
return ret;
}
int ObMViewSchedJobUtils::resolve_date_expr_to_timestamp(
ObResolverParams &params,
ObSQLSessionInfo &session,
const ParseNode &node,
ObIAllocator &allocator,
int64_t &timestamp)
{
int ret = OB_SUCCESS;
ObRawExpr *const_expr = nullptr;
ParamStore params_array;
ObObj obj;
timestamp = 0;
if (OB_FAIL(ObResolverUtils::resolve_const_expr(
params, node, const_expr, nullptr))) {
LOG_WARN("fail to resolve const expr", KR(ret));
} else if (OB_FAIL(ObSQLUtils::calc_const_expr(
&session, *const_expr, obj, allocator, params_array))) {
LOG_WARN("fail to calc const expr", KR(ret));
} else {
if (obj.is_timestamp()) {
timestamp = obj.get_timestamp();
} else if (obj.is_timestamp_tz()) {
timestamp = obj.get_otimestamp_value().time_us_;
} else if (obj.is_datetime()) {
int64_t tmp_timestamp = 0;
if (OB_FAIL(ObTimeConverter::datetime_to_timestamp(
obj.get_datetime(), TZ_INFO(&session), tmp_timestamp))) {
LOG_WARN("failed to convert datetime to timestamp",
KR(ret), K(obj.get_datetime()));
} else {
timestamp = tmp_timestamp;
}
} else {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected obj type", KR(ret), K(obj));
}
}
return ret;
}
} // end of storage
} // end of oceanbase

17
src/storage/mview/ob_mview_sched_job_utils.h
View File

@ -13,6 +13,7 @@
#pragma once
#include "lib/ob_define.h"
#include "sql/parser/parse_node.h"
namespace oceanbase
{
@ -34,6 +35,7 @@ class ObString;
}
namespace sql
{
class ObResolverParams;
class ObSQLSessionInfo;
}
namespace dbms_scheduler
@ -84,16 +86,19 @@ public:
const uint64_t tenant_id,
const uint64_t table_id);
static int calc_date_expression_from_str(sql::ObSQLSessionInfo &session,
static int calc_date_expr_from_str(sql::ObSQLSessionInfo &session,
common::ObIAllocator &allocator,
const uint64_t tenant_id,
const ObString &date_str,
common::ObObj &date_obj);
static int convert_session_date_to_utc(sql::ObSQLSessionInfo *session_info,
const common::ObObj &in_obj,
common::ObObj &out_obj);
int64_t &timestamp);
static int calc_date_expression(dbms_scheduler::ObDBMSSchedJobInfo &job_info,
int64_t &next_date_utc_ts);
int64_t &next_date_ts);
static int resolve_date_expr_to_timestamp(sql::ObResolverParams &params,
sql::ObSQLSessionInfo &session,
const ParseNode &node,
common::ObIAllocator &allocator,
int64_t &timestamp);
public:
static constexpr int64_t JOB_ID_OFFSET = 1000000L;