fix bug about tenant_time_zone when switch tennat
This commit is contained in:
obdev
ob-robot
parent
2fc9f4a99d
commit
780b56301a
@ -39013,7 +39013,7 @@ ObTZNameKey::ObTZNameKey(const ObString &tz_key_str)
|
||||
{
|
||||
int64_t len = tz_key_str.length();
|
||||
if (OB_UNLIKELY(len + 1 > OB_MAX_TZ_NAME_LEN)) {
|
||||
LOG_ERROR_RET(OB_INVALID_ARGUMENT, "invalid tz_key_str", K(tz_key_str));
|
||||
LOG_WARN_RET(OB_INVALID_ARGUMENT, "invalid tz_key_str", K(tz_key_str));
|
||||
} else {
|
||||
for (int64_t i = 0; i < len; ++i) {
|
||||
tz_name_[i] = static_cast<char>(tolower(tz_key_str[i]));
|
||||
|
||||
@ -142,43 +142,41 @@ void ObTimeZoneInfoManager::TaskProcessThread::handle(void *task)
|
||||
int ObTimeZoneInfoManager::fetch_time_zone_info()
|
||||
{
|
||||
int ret = OB_SUCCESS;
|
||||
MTL_SWITCH(tenant_id_) {
|
||||
if (OB_UNLIKELY(!inited_)) {
|
||||
ret = OB_NOT_INIT;
|
||||
LOG_WARN("not init", K(ret));
|
||||
} else {
|
||||
int64_t current_tz_version = -1;
|
||||
ObSQLClientRetryWeak sql_client_retry_weak(&sql_proxy_, tenant_id_, OB_ALL_SYS_STAT_TID);
|
||||
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
|
||||
sqlclient::ObMySQLResult *result = NULL;
|
||||
if (OB_FAIL(sql_client_retry_weak.read(res, tenant_id_, FETCH_LATEST_TZ_VERSION_SQL))) {
|
||||
LOG_WARN("fail to execute sql", K(ret), K(tenant_id_));
|
||||
} else if (OB_ISNULL(result = res.get_result())) {
|
||||
ret = OB_ERR_UNEXPECTED;
|
||||
LOG_WARN("fail to get result", K(result), K(ret));
|
||||
} else if (OB_FAIL(result->next())) {
|
||||
if (OB_ITER_END == ret) {
|
||||
// all_sys_stat中没有timezone_version，说明处于升级过程中
|
||||
ret = OB_SUCCESS;
|
||||
} else {
|
||||
LOG_WARN("ObMySQLResult next failed", K(ret));
|
||||
}
|
||||
if (OB_UNLIKELY(!inited_)) {
|
||||
ret = OB_NOT_INIT;
|
||||
LOG_WARN("not init", K(ret));
|
||||
} else {
|
||||
int64_t current_tz_version = -1;
|
||||
ObSQLClientRetryWeak sql_client_retry_weak(&sql_proxy_, tenant_id_, OB_ALL_SYS_STAT_TID);
|
||||
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
|
||||
sqlclient::ObMySQLResult *result = NULL;
|
||||
if (OB_FAIL(sql_client_retry_weak.read(res, tenant_id_, FETCH_LATEST_TZ_VERSION_SQL))) {
|
||||
LOG_WARN("fail to execute sql", K(ret), K(tenant_id_));
|
||||
} else if (OB_ISNULL(result = res.get_result())) {
|
||||
ret = OB_ERR_UNEXPECTED;
|
||||
LOG_WARN("fail to get result", K(result), K(ret));
|
||||
} else if (OB_FAIL(result->next())) {
|
||||
if (OB_ITER_END == ret) {
|
||||
// all_sys_stat中没有timezone_version，说明处于升级过程中
|
||||
ret = OB_SUCCESS;
|
||||
} else {
|
||||
ObString version_str;
|
||||
EXTRACT_VARCHAR_FIELD_MYSQL(*result, "value", version_str);
|
||||
bool is_valid = false;
|
||||
current_tz_version = ObFastAtoi<int64_t>::atoi(version_str.ptr(),
|
||||
version_str.ptr() + version_str.length(), is_valid);
|
||||
if (!is_valid) {
|
||||
ret = OB_ERR_UNEXPECTED;
|
||||
LOG_WARN("invalid key version", K(ret), K(version_str));
|
||||
}
|
||||
LOG_WARN("ObMySQLResult next failed", K(ret));
|
||||
}
|
||||
} else {
|
||||
ObString version_str;
|
||||
EXTRACT_VARCHAR_FIELD_MYSQL(*result, "value", version_str);
|
||||
bool is_valid = false;
|
||||
current_tz_version = ObFastAtoi<int64_t>::atoi(version_str.ptr(),
|
||||
version_str.ptr() + version_str.length(), is_valid);
|
||||
if (!is_valid) {
|
||||
ret = OB_ERR_UNEXPECTED;
|
||||
LOG_WARN("invalid key version", K(ret), K(version_str));
|
||||
}
|
||||
}
|
||||
if (OB_FAIL(ret)) {
|
||||
} else if (OB_FAIL(fetch_time_zone_info_from_tenant_table(current_tz_version))) {
|
||||
LOG_WARN("fetch timezone info from tenant tz table failed", K(ret));
|
||||
}
|
||||
}
|
||||
if (OB_FAIL(ret)) {
|
||||
} else if (OB_FAIL(fetch_time_zone_info_from_tenant_table(current_tz_version))) {
|
||||
LOG_WARN("fetch timezone info from tenant tz table failed", K(ret));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -26,7 +26,7 @@ class TimeZoneInfoImporter:
|
||||
def generate_sql(self):
|
||||
self.sql_list = []
|
||||
self.tz_version_sql_list = []
|
||||
self.expect_count = [0, 0, 0, 0];
|
||||
self.expect_count = [0, 0, 0, 0]
|
||||
replace_str1 = 'TRUNCATE TABLE time_zone'
|
||||
replace_str2 = 'INTO time_zone'
|
||||
replace_str3 = 'ALTER TABLE time_zone_transition'
|
||||
@ -35,7 +35,7 @@ class TimeZoneInfoImporter:
|
||||
replace_count_str2 = 'time_zone_transition count:'
|
||||
replace_count_str3 = 'time_zone_transition_type count:'
|
||||
with open(self.file_name) as f_read:
|
||||
sql = "";
|
||||
sql = ""
|
||||
for line in f_read:
|
||||
if re.search('__all_sys_stat', line, re.IGNORECASE):
|
||||
self.tz_version_sql_list.append(line)
|
||||
@ -61,7 +61,7 @@ class TimeZoneInfoImporter:
|
||||
sql += new_line
|
||||
if ";" in new_line:
|
||||
self.sql_list.append(sql)
|
||||
sql = "";
|
||||
sql = ""
|
||||
|
||||
def connect_server(self):
|
||||
self.conn = mysql.connector.connect(user='root', password=self.pwd, host=self.host, port=self.port, database='mysql')
|
||||
@ -92,57 +92,61 @@ class TimeZoneInfoImporter:
|
||||
self.cur.execute(sql)
|
||||
except mysql.connector.Error as err:
|
||||
print("ERROR : " + sql)
|
||||
print(err);
|
||||
print(err)
|
||||
raise
|
||||
|
||||
def execute_sql(self):
|
||||
try:
|
||||
for sql in self.sql_list:
|
||||
self.cur.execute(sql);
|
||||
print "INFO : execute sql -- {0}".format(sql)
|
||||
self.cur.execute("")
|
||||
except mysql.connector.Error as err:
|
||||
print("ERROR : " + sql)
|
||||
print(err);
|
||||
print("ERROR : fail to import time zone info");
|
||||
print(err)
|
||||
print("ERROR : fail to import time zone info")
|
||||
raise
|
||||
else:
|
||||
print("INFO : success to import time zone info");
|
||||
print("INFO : success to import time zone info")
|
||||
|
||||
def execute_check_sql(self, table_name, idx):
|
||||
self.cur.execute("select count(*) from {0}".format(table_name));
|
||||
result = self.cur.fetchone();
|
||||
self.result_count[idx] = result[0];
|
||||
print "INFO : {0} record count -- {1}, expect count -- {2}".format(table_name, result[0], self.expect_count[idx]);
|
||||
self.cur.execute("select count(*) from {0}".format(table_name))
|
||||
result = self.cur.fetchone()
|
||||
self.result_count[idx] = result[0]
|
||||
print "INFO : {0} record count -- {1}, expect count -- {2}".format(table_name, result[0], self.expect_count[idx])
|
||||
|
||||
def check_result(self):
|
||||
self.result_count = [0, 0, 0, 0];
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone", 0);
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_name", 1);
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_transition", 2);
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_transition_type", 3);
|
||||
self.result_count = [0, 0, 0, 0]
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone", 0)
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_name", 1)
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_transition", 2)
|
||||
self.execute_check_sql("oceanbase.__all_tenant_time_zone_transition_type", 3)
|
||||
if self.expect_count[0] == self.result_count[0] \
|
||||
and self.expect_count[1] == self.result_count[1] \
|
||||
and self.expect_count[2] == self.result_count[2] \
|
||||
and self.expect_count[3] == self.result_count[3]:
|
||||
try:
|
||||
for sql in self.tz_version_sql_list:
|
||||
self.cur.execute(sql);
|
||||
self.cur.execute(sql)
|
||||
print "INFO : execute sql -- {0}".format(sql)
|
||||
self.cur.execute("")
|
||||
except mysql.connector.Error as err:
|
||||
print("ERROR : " + sql)
|
||||
print(err);
|
||||
print("ERROR : fail to insert time zone version");
|
||||
print(err)
|
||||
print("ERROR : fail to insert time zone version")
|
||||
raise
|
||||
else:
|
||||
print("INFO : success to insert time zone version");
|
||||
print("INFO : success to insert time zone version")
|
||||
|
||||
def main():
|
||||
tz_info_importer = TimeZoneInfoImporter();
|
||||
tz_info_importer.get_args();
|
||||
tz_info_importer.connect_server();
|
||||
if False == tz_info_importer.upgrade_mode:
|
||||
tz_info_importer.generate_sql();
|
||||
tz_info_importer.execute_sql();
|
||||
tz_info_importer.check_result();
|
||||
tz_info_importer = TimeZoneInfoImporter()
|
||||
tz_info_importer.get_args()
|
||||
try:
|
||||
tz_info_importer.connect_server()
|
||||
if False == tz_info_importer.upgrade_mode:
|
||||
tz_info_importer.generate_sql()
|
||||
tz_info_importer.execute_sql()
|
||||
tz_info_importer.check_result()
|
||||
except:
|
||||
print("except error in main")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user