forked from huawei/openGauss-server
Merge pull request 'dbmind中sql语句执行文件的注释' (#11) from Eukanj827/openGauss-server:master into master
This commit is contained in:
commit
4b7af41aff
|
|
@ -18,13 +18,19 @@ import psycopg2
|
||||||
from .execute_factory import ExecuteFactory
|
from .execute_factory import ExecuteFactory
|
||||||
from .execute_factory import IndexInfo
|
from .execute_factory import IndexInfo
|
||||||
|
|
||||||
|
#class name: DriverExecute (Inherits from the parent class ExecuteFactory)
|
||||||
|
#description: The SQL statement performs the operations associated with the call
|
||||||
|
#date: 2022/8/10
|
||||||
|
#contact: 1865997821
|
||||||
|
|
||||||
class DriverExecute(ExecuteFactory):
|
class DriverExecute(ExecuteFactory):
|
||||||
def __init__(self, *arg):
|
def __init__(self, *arg):
|
||||||
|
#Call the arguments of the parent class __init__ method
|
||||||
super(DriverExecute, self).__init__(*arg)
|
super(DriverExecute, self).__init__(*arg)
|
||||||
self.conn = None
|
self.conn = None
|
||||||
self.cur = None
|
self.cur = None
|
||||||
|
|
||||||
|
#Connecting to the database
|
||||||
def init_conn_handle(self):
|
def init_conn_handle(self):
|
||||||
self.conn = psycopg2.connect(dbname=self.dbname,
|
self.conn = psycopg2.connect(dbname=self.dbname,
|
||||||
user=self.user,
|
user=self.user,
|
||||||
|
|
@ -33,6 +39,7 @@ class DriverExecute(ExecuteFactory):
|
||||||
port=self.port)
|
port=self.port)
|
||||||
self.cur = self.conn.cursor()
|
self.cur = self.conn.cursor()
|
||||||
|
|
||||||
|
#If an error occurs after the SQL statement is executed, the error information is reported to the user
|
||||||
def execute(self, sql):
|
def execute(self, sql):
|
||||||
try:
|
try:
|
||||||
self.cur.execute(sql)
|
self.cur.execute(sql)
|
||||||
|
|
@ -41,11 +48,13 @@ class DriverExecute(ExecuteFactory):
|
||||||
except Exception:
|
except Exception:
|
||||||
self.conn.commit()
|
self.conn.commit()
|
||||||
|
|
||||||
|
#Disconnecting from the database
|
||||||
def close_conn(self):
|
def close_conn(self):
|
||||||
if self.conn and self.cur:
|
if self.conn and self.cur:
|
||||||
self.cur.close()
|
self.cur.close()
|
||||||
self.conn.close()
|
self.conn.close()
|
||||||
|
|
||||||
|
#Check whether multiple nodes exist
|
||||||
def is_multi_node(self):
|
def is_multi_node(self):
|
||||||
self.init_conn_handle()
|
self.init_conn_handle()
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,11 @@
|
||||||
|
|
||||||
import re
|
import re
|
||||||
|
|
||||||
|
#class name: IndexInfo
|
||||||
|
#description: Define information about table indexes
|
||||||
|
#methods: __init__
|
||||||
|
#date: 2022/8/10
|
||||||
|
#contact: 1865997821
|
||||||
|
|
||||||
class IndexInfo:
|
class IndexInfo:
|
||||||
def __init__(self, schema, table, indexname, columns, indexdef):
|
def __init__(self, schema, table, indexname, columns, indexdef):
|
||||||
|
|
@ -24,7 +29,9 @@ class IndexInfo:
|
||||||
self.primary_key = False
|
self.primary_key = False
|
||||||
self.redundant_obj = []
|
self.redundant_obj = []
|
||||||
|
|
||||||
|
#class name: ExecuteFactory
|
||||||
|
#date: 2022/8/10
|
||||||
|
#contact: 1865997821
|
||||||
class ExecuteFactory:
|
class ExecuteFactory:
|
||||||
def __init__(self, dbname, user, password, host, port, schema, multi_node, max_index_storage):
|
def __init__(self, dbname, user, password, host, port, schema, multi_node, max_index_storage):
|
||||||
self.dbname = dbname
|
self.dbname = dbname
|
||||||
|
|
@ -36,11 +43,11 @@ class ExecuteFactory:
|
||||||
self.max_index_storage = max_index_storage
|
self.max_index_storage = max_index_storage
|
||||||
self.multi_node = multi_node
|
self.multi_node = multi_node
|
||||||
|
|
||||||
|
# Record redundant indexes
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def record_redundant_indexes(cur_table_indexes, redundant_indexes):
|
def record_redundant_indexes(cur_table_indexes, redundant_indexes):
|
||||||
cur_table_indexes = sorted(cur_table_indexes,
|
cur_table_indexes = sorted(cur_table_indexes,
|
||||||
key=lambda index_obj: len(index_obj.columns.split(',')))
|
key=lambda index_obj: len(index_obj.columns.split(',')))
|
||||||
# record redundant indexes
|
|
||||||
for pos, index in enumerate(cur_table_indexes[:-1]):
|
for pos, index in enumerate(cur_table_indexes[:-1]):
|
||||||
is_redundant = False
|
is_redundant = False
|
||||||
for candidate_index in cur_table_indexes[pos + 1:]:
|
for candidate_index in cur_table_indexes[pos + 1:]:
|
||||||
|
|
@ -52,6 +59,7 @@ class ExecuteFactory:
|
||||||
if is_redundant:
|
if is_redundant:
|
||||||
redundant_indexes.append(index)
|
redundant_indexes.append(index)
|
||||||
|
|
||||||
|
#Match the name of the table against the index of the query
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def match_table_name(table_name, query_index_dict):
|
def match_table_name(table_name, query_index_dict):
|
||||||
for elem in query_index_dict.keys():
|
for elem in query_index_dict.keys():
|
||||||
|
|
@ -66,6 +74,7 @@ class ExecuteFactory:
|
||||||
return False, table_name
|
return False, table_name
|
||||||
return True, table_name
|
return True, table_name
|
||||||
|
|
||||||
|
#Retrieves a valid index based on the regular expression, adding the corresponding index and empty element if none exists
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_valid_indexes(record, hypoid_table_column, valid_indexes):
|
def get_valid_indexes(record, hypoid_table_column, valid_indexes):
|
||||||
tokens = record.split(' ')
|
tokens = record.split(' ')
|
||||||
|
|
@ -88,6 +97,7 @@ class ExecuteFactory:
|
||||||
if columns not in valid_indexes[table_name]:
|
if columns not in valid_indexes[table_name]:
|
||||||
valid_indexes[table_name].append((columns, index_type))
|
valid_indexes[table_name].append((columns, index_type))
|
||||||
|
|
||||||
|
#Record invalid SQL statements and returns the corresponding help information that matches the corresponding SQL statement
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def record_ineffective_negative_sql(candidate_index, obj, ind):
|
def record_ineffective_negative_sql(candidate_index, obj, ind):
|
||||||
cur_table = candidate_index.table
|
cur_table = candidate_index.table
|
||||||
|
|
@ -125,6 +135,7 @@ class ExecuteFactory:
|
||||||
candidate_index.ineffective_pos.append(ind)
|
candidate_index.ineffective_pos.append(ind)
|
||||||
candidate_index.total_sql_num += obj.frequency
|
candidate_index.total_sql_num += obj.frequency
|
||||||
|
|
||||||
|
#Returns the last input and the corresponding result
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def match_last_result(table_name, index_column, history_indexes, history_invalid_indexes):
|
def match_last_result(table_name, index_column, history_indexes, history_invalid_indexes):
|
||||||
for column in history_indexes.get(table_name, dict()):
|
for column in history_indexes.get(table_name, dict()):
|
||||||
|
|
@ -142,6 +153,7 @@ class ExecuteFactory:
|
||||||
if not history_indexes[table_name]:
|
if not history_indexes[table_name]:
|
||||||
del history_indexes[table_name]
|
del history_indexes[table_name]
|
||||||
|
|
||||||
|
#Correcting SQL statements
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def make_single_advisor_sql(ori_sql):
|
def make_single_advisor_sql(ori_sql):
|
||||||
sql = 'select gs_index_advise(\''
|
sql = 'select gs_index_advise(\''
|
||||||
|
|
|
||||||
|
|
@ -23,12 +23,16 @@ from .execute_factory import IndexInfo
|
||||||
|
|
||||||
BASE_CMD = None
|
BASE_CMD = None
|
||||||
|
|
||||||
|
#class name: GSqlExecute
|
||||||
|
#description: Solve the optimization problem of GSQL statement execution
|
||||||
|
#date: 2022/8/11
|
||||||
|
#contact: 1865997821
|
||||||
class GSqlExecute(ExecuteFactory):
|
class GSqlExecute(ExecuteFactory):
|
||||||
def __init__(self, *args):
|
def __init__(self, *args):
|
||||||
super(GSqlExecute, self).__init__(*args)
|
super(GSqlExecute, self).__init__(*args)
|
||||||
|
|
||||||
def init_conn_handle(self):
|
def init_conn_handle(self):
|
||||||
|
#define a global variable BASE_CMD,it is a connection command statement
|
||||||
global BASE_CMD
|
global BASE_CMD
|
||||||
BASE_CMD = 'gsql -p ' + str(self.port) + ' -d ' + self.dbname
|
BASE_CMD = 'gsql -p ' + str(self.port) + ' -d ' + self.dbname
|
||||||
if self.host:
|
if self.host:
|
||||||
|
|
@ -38,6 +42,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
if self.password:
|
if self.password:
|
||||||
BASE_CMD += ' -W ' + self.password
|
BASE_CMD += ' -W ' + self.password
|
||||||
|
|
||||||
|
#Run the shell command in BASE_CMD
|
||||||
def run_shell_cmd(self, target_sql_list):
|
def run_shell_cmd(self, target_sql_list):
|
||||||
cmd = BASE_CMD + ' -c \"'
|
cmd = BASE_CMD + ' -c \"'
|
||||||
if self.schema:
|
if self.schema:
|
||||||
|
|
@ -47,6 +52,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
cmd += '\"'
|
cmd += '\"'
|
||||||
proc = subprocess.Popen(
|
proc = subprocess.Popen(
|
||||||
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
|
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
|
||||||
|
#Read data from stdout and stderr,If an error message is displayed, an error message is displayed
|
||||||
(stdout, stderr) = proc.communicate()
|
(stdout, stderr) = proc.communicate()
|
||||||
stdout, stderr = stdout.decode(), stderr.decode()
|
stdout, stderr = stdout.decode(), stderr.decode()
|
||||||
if 'gsql: FATAL:' in stderr or 'failed to connect' in stderr:
|
if 'gsql: FATAL:' in stderr or 'failed to connect' in stderr:
|
||||||
|
|
@ -74,6 +80,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
print(e.output.decode(), file=sys.stderr)
|
print(e.output.decode(), file=sys.stderr)
|
||||||
return int(ret.decode().strip().split()[2]) > 0
|
return int(ret.decode().strip().split()[2]) > 0
|
||||||
|
|
||||||
|
#Parse the recommended result returned
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def parse_single_advisor_result(res, table_index_dict):
|
def parse_single_advisor_result(res, table_index_dict):
|
||||||
if len(res) > 2 and res[0:2] == ' (':
|
if len(res) > 2 and res[0:2] == ' (':
|
||||||
|
|
@ -183,6 +190,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
total_cost = 0
|
total_cost = 0
|
||||||
found_plan = False
|
found_plan = False
|
||||||
hypo_index = False
|
hypo_index = False
|
||||||
|
# create hypo-indexes
|
||||||
for line in res:
|
for line in res:
|
||||||
if 'QUERY PLAN' in line:
|
if 'QUERY PLAN' in line:
|
||||||
found_plan = True
|
found_plan = True
|
||||||
|
|
@ -222,6 +230,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
i += 1
|
i += 1
|
||||||
return total_cost
|
return total_cost
|
||||||
|
|
||||||
|
#Production workflows consume report files
|
||||||
def estimate_workload_cost_file(self, workload, index_config=None, ori_indexes_name=None):
|
def estimate_workload_cost_file(self, workload, index_config=None, ori_indexes_name=None):
|
||||||
sql_file = str(time.time()) + '.sql'
|
sql_file = str(time.time()) + '.sql'
|
||||||
is_computed = False
|
is_computed = False
|
||||||
|
|
@ -264,6 +273,7 @@ class GSqlExecute(ExecuteFactory):
|
||||||
|
|
||||||
return total_cost
|
return total_cost
|
||||||
|
|
||||||
|
#Check for empty indexes and note them to optimize the table structure
|
||||||
def check_useless_index(self, history_indexes, history_invalid_indexes):
|
def check_useless_index(self, history_indexes, history_invalid_indexes):
|
||||||
schemas = [elem.lower()
|
schemas = [elem.lower()
|
||||||
for elem in filter(None, self.schema.split(','))]
|
for elem in filter(None, self.schema.split(','))]
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue