Charl
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Index recommendation function enhancement

This commit is contained in:
flyly 2021-07-28 20:04:26 +08:00
parent 6c642dfb0f
commit b8b38213df
4 changed files with 380 additions and 296 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
src/gausskernel/dbmind/tools/index_advisor/extract_log.py
View File

@ -1,4 +1,3 @@
import re
import os
import argparse
@ -36,12 +35,14 @@ def get_workload_template(templates, sqls):
def output_valid_sql(sql):
if 'from pg_' in sql.lower() or ' join ' in sql.lower():
is_quotation_valid = sql.count("'") % 2
if 'from pg_' in sql.lower() or ' join ' in sql.lower() or is_quotation_valid:
return ''
if any(tp in sql.lower() for tp in SQL_TYPE[1:]):
return sql if sql.endswith('; ') else sql + ';'
sql = re.sub(r'for\s+update[\s;]*$', '', sql, flags=re.I)
return sql.strip() if sql.endswith('; ') else sql + ';'
elif SQL_TYPE[0] in sql.lower() and 'from ' in sql.lower():
return sql if sql.endswith('; ') else sql + ';'
return sql.strip() if sql.endswith('; ') else sql + ';'
return ''
@ -103,9 +104,11 @@ def get_parsed_sql(file, user, database, sql_amount, statement):
reverse=True)
for item in param_list:
if len(item[1].strip()) >= 256:
sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else ('$' + item[0].strip()), "''")
sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else
('$' + item[0].strip()), "''")
else:
sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else ('$' + item[0].strip()), item[1].strip())
sql = sql.replace(item[0].strip() if re.match(r'\$', item[0]) else
('$' + item[0].strip()), item[1].strip())
if output_valid_sql(sql):
SQL_AMOUNT += 1
yield output_valid_sql(sql)
@ -132,34 +135,47 @@ def get_start_position(start_time, file_path):
return -1
def extract_sql_from_log(log_path, output_path, database, user, start_time, sql_amount, statement):
templates = {}
files = os.listdir(log_path)
files = sorted(files, key=lambda x: os.path.getctime(os.path.join(log_path, x)), reverse=True)
valid_files = files
time_stamp = int(time.mktime(time.strptime(start_time, '%Y-%m-%d %H:%M:%S')))
if start_time:
valid_files = []
for file in files:
if os.path.getmtime(os.path.join(log_path, file)) < time_stamp:
break
valid_files.insert(0, file)
def record_sql(valid_files, args, output_obj):
for ind, file in enumerate(valid_files):
if sql_amount and SQL_AMOUNT >= sql_amount:
if args.sql_amount and SQL_AMOUNT >= args.sql_amount:
break
file_path = os.path.join(log_path, file)
file_path = os.path.join(args.l, file)
if os.path.isfile(file_path) and re.search(r'.log$', file):
start_position = 0
if ind == 0:
start_position = get_start_position(start_time, file_path)
start_position = get_start_position(args.start_time, file_path)
if start_position == -1:
continue
with open(file_path, mode='r') as f:
f.seek(start_position, 0)
get_workload_template(templates, get_parsed_sql(f, user, database, sql_amount,
statement))
with open(output_path, 'w') as output_file:
json.dump(templates, output_file)
if isinstance(output_obj, dict):
get_workload_template(output_obj, get_parsed_sql(f, args.U, args.d,
args.sql_amount,
args.statement))
else:
for sql in get_parsed_sql(f, args.U, args.d, args.sql_amount, args.statement):
output_obj.write(sql + '\n')
def extract_sql_from_log(args):
files = os.listdir(args.l)
files = sorted(files, key=lambda x: os.path.getctime(os.path.join(args.l, x)), reverse=True)
valid_files = files
time_stamp = int(time.mktime(time.strptime(args.start_time, '%Y-%m-%d %H:%M:%S')))
if args.start_time:
valid_files = []
for file in files:
if os.path.getmtime(os.path.join(args.l, file)) < time_stamp:
break
valid_files.insert(0, file)
if args.json:
templates = {}
record_sql(valid_files, args, templates)
with open(args.f, 'w') as output_file:
json.dump(templates, output_file)
else:
with open(args.f, 'w') as output_file:
record_sql(valid_files, args, output_file)
def main():
@ -172,17 +188,17 @@ def main():
arg_parser.add_argument("--sql_amount", help="The number of sql collected", type=int)
arg_parser.add_argument("--statement", action='store_true', help="Extract statement log type",
default=False)
arg_parser.add_argument("--json", action='store_true',
help="Whether the workload file format is json", default=False)
args = arg_parser.parse_args()
if args.start_time:
time.strptime(args.start_time, '%Y-%m-%d %H:%M:%S')
if args.sql_amount and args.sql_amount <= 0:
raise argparse.ArgumentTypeError("%s is an invalid positive int value" % args.sql_amount)
extract_sql_from_log(args.l, args.f, args.d, args.U, args.start_time, args.sql_amount,
args.statement)
extract_sql_from_log(args)
if __name__ == '__main__':
main()

174
src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload.py
View File

@ -107,6 +107,7 @@ class IndexInfo:
self.indexname = indexname
self.columns = columns
self.indexdef = indexdef
self.primary_key = False
self.redundant_obj = []
@ -164,7 +165,10 @@ def filter_low_benefit(pos_list, candidate_indexes, multi_iter_mode, workload):
sql_optimzed += 1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0]
negative_ratio = (index.insert_sql_num + index.delete_sql_num + index.update_sql_num) / \
index.total_sql_num
if sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio:
# filter the candidate indexes that do not meet the conditions of optimization
if sql_optimzed / len(index.positive_pos) < 0.1:
remove_list.append(key)
elif sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio:
remove_list.append(key)
for item in sorted(remove_list, reverse=True):
candidate_indexes.pop(item)
@ -200,6 +204,9 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total,
sql_info = {'sqlDetails': []}
benefit_types = [index.ineffective_pos, index.positive_pos, index.negative_pos]
for category, benefit_type in enumerate(benefit_types):
sql_count = 0
for item in benefit_type:
sql_count += workload[item].frequency
for ind, pos in enumerate(benefit_type):
sql_detail = {}
sql_template = workload[pos].statement
@ -208,23 +215,26 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total,
sql_detail['sqlTemplate'] = sql_template
sql_detail['sql'] = workload[pos].statement
sql_detail['sqlCount'] = sql_count
if category == 1:
sql_optimzed = (1 - workload[pos].cost_list[cost_list_pos] /
workload[pos].cost_list[0])
sql_detail['optimized'] = '%.2f' % (sql_optimzed * 100)
sql_optimzed = (workload[pos].cost_list[0] -
workload[pos].cost_list[cost_list_pos]) / \
workload[pos].cost_list[cost_list_pos]
sql_detail['optimized'] = '%.3f' % sql_optimzed
sql_detail['correlationType'] = category
sql_info['sqlDetails'].append(sql_detail)
workload_optimized = 1 - index_cost_total[cost_list_pos] / index_cost_total[0]
sql_info['workloadOptimized'] = '%.2f' % (workload_optimized * 100)
workload_optimized = (1 - index_cost_total[cost_list_pos] / index_cost_total[0]) * 100
sql_info['workloadOptimized'] = '%.2f' % (workload_optimized if workload_optimized > 1 else 1)
sql_info['schemaName'] = SCHEMA
sql_info['tbName'] = table_name
sql_info['columns'] = index.columns
sql_info['statement'] = statement
sql_info['dmlCount'] = round(index.total_sql_num)
sql_info['selectRatio'] = round(index.select_sql_num / index.total_sql_num, 2)
sql_info['insertRatio'] = round(index.insert_sql_num / index.total_sql_num, 2)
sql_info['deleteRatio'] = round(index.delete_sql_num / index.total_sql_num, 2)
sql_info['updateRatio'] = round(index.update_sql_num / index.total_sql_num, 2)
sql_info['selectRatio'] = round(index.select_sql_num * 100 / index.total_sql_num, 2)
sql_info['insertRatio'] = round(index.insert_sql_num * 100 / index.total_sql_num, 2)
sql_info['deleteRatio'] = round(index.delete_sql_num * 100 / index.total_sql_num, 2)
sql_info['updateRatio'] = round(100 - sql_info['selectRatio'] - sql_info['insertRatio']
- sql_info['deleteRatio'], 2)
display_info['recommendIndexes'].append(sql_info)
return display_info
@ -233,13 +243,25 @@ def record_redundant_indexes(cur_table_indexes, redundant_indexes):
cur_table_indexes = sorted(cur_table_indexes,
key=lambda index_obj: len(index_obj.columns.split(',')))
# record redundant indexes
has_restore = []
for pos, index in enumerate(cur_table_indexes[:-1]):
is_redundant = False
for candidate_index in cur_table_indexes[pos + 1:]:
if 'UNIQUE INDEX' in index.indexdef:
# ensure that UNIQUE INDEX will not become redundant compared to normal index
if 'UNIQUE INDEX' not in candidate_index.indexdef:
continue
# ensure redundant index not is pkey
elif index.primary_key:
if re.match(r'%s' % candidate_index.columns, index.columns):
candidate_index.redundant_obj.append(index)
redundant_indexes.append(candidate_index)
has_restore.append(candidate_index)
continue
if re.match(r'%s' % index.columns, candidate_index.columns):
is_redundant = True
index.redundant_obj.append(candidate_index)
if is_redundant:
if is_redundant and index not in has_restore:
redundant_indexes.append(index)
@ -249,8 +271,14 @@ def check_useless_index(tables):
if not tables:
return whole_indexes, redundant_indexes
tables_string = ','.join(["'%s'" % table for table in tables[SCHEMA]])
sql = "select tablename, indexname, indexdef from pg_indexes where " \
"schemaname='%s' and tablename in (%s) order by tablename " % (SCHEMA, tables_string)
sql = "SELECT c.relname AS tablename, i.relname AS indexname, " \
"pg_get_indexdef(i.oid) AS indexdef, p.contype AS pkey from " \
"pg_index x JOIN pg_class c ON c.oid = x.indrelid JOIN " \
"pg_class i ON i.oid = x.indexrelid LEFT JOIN pg_namespace n " \
"ON n.oid = c.relnamespace LEFT JOIN pg_constraint p ON i.oid = p.conindid" \
"WHERE (c.relkind = ANY (ARRAY['r'::\"char\", 'm'::\"char\"])) AND " \
"(i.relkind = ANY (ARRAY['i'::\"char\", 'I'::\"char\"])) AND " \
"n.nspname = '%s' AND c.relname in (%s) order by c.relname;" % (SCHEMA, tables_string)
res = run_shell_cmd([sql]).split('\n')
if res:
@ -261,9 +289,11 @@ def check_useless_index(tables):
elif re.match(r'\(\d+ rows?\)', line):
continue
elif '|' in line:
table, index, indexdef = [item.strip() for item in line.split('|')]
cur_columns = re.search(r'\((.*)\)', indexdef).group(1)
table, index, indexdef, pkey = [item.strip() for item in line.split('|')]
cur_columns = re.search(r'\(([^\(\)]*)\)', indexdef).group(1)
cur_index_obj = IndexInfo(SCHEMA, table, index, cur_columns, indexdef)
if pkey:
cur_index_obj.primary_key = True
whole_indexes.append(cur_index_obj)
if cur_table_indexes and cur_table_indexes[-1].table != table:
record_redundant_indexes(cur_table_indexes, redundant_indexes)
@ -293,30 +323,39 @@ def check_unused_index_workload(whole_indexes, redundant_indexes, workload_index
indexes_name = set(index.indexname for index in whole_indexes)
unused_index = list(indexes_name.difference(workload_indexes))
remove_list = []
for pos, index in enumerate(redundant_indexes):
is_redundant = False
for redundant_obj in index.redundant_obj:
if redundant_obj.indexname not in unused_index:
is_redundant = True
if not is_redundant:
remove_list.append(pos)
for item in sorted(remove_list, reverse=True):
redundant_indexes.pop(item)
print_header_boundary(" Current workload useless indexes ")
if not unused_index:
print("No useless index!")
detail_info['uselessIndexes'] = []
# useless index
unused_index_columns = dict()
for cur_index in unused_index:
if not re.search('_pkey$', cur_index):
for index in whole_indexes:
if cur_index == index.indexname:
for index in whole_indexes:
if cur_index == index.indexname:
unused_index_columns[cur_index] = index.columns
if 'UNIQUE INDEX' not in index.indexdef:
statement = "DROP INDEX %s;" % index.indexname
print(statement)
useless_index = {"schemaName": index.schema, "tbName": index.table, "type": 1,
"columns": index.columns, "statement": statement}
detail_info['uselessIndexes'].append(useless_index)
print_header_boundary(" Redundant indexes ")
# filter redundant index
for pos, index in enumerate(redundant_indexes):
is_redundant = False
for redundant_obj in index.redundant_obj:
# redundant objects are not in the useless index set or
# equal to the column value in the useless index must be redundant index
index_exist = redundant_obj.indexname not in unused_index_columns.keys() or \
(unused_index_columns.get(redundant_obj.indexname) and
redundant_obj.columns == unused_index_columns[redundant_obj.indexname])
if index_exist:
is_redundant = True
if not is_redundant:
remove_list.append(pos)
for item in sorted(remove_list, reverse=True):
redundant_indexes.pop(item)
if not redundant_indexes:
print("No redundant index!")
# redundant index
@ -378,6 +417,7 @@ def get_workload_template(workload):
def workload_compression(input_path):
compressed_workload = []
total_num = 0
if JSON_TYPE:
with open(input_path, 'r') as file:
templates = json.load(file)
@ -389,7 +429,8 @@ def workload_compression(input_path):
for sql in elem['samples']:
compressed_workload.append(QueryItem(sql.strip('\n'),
elem['cnt'] / len(elem['samples'])))
return compressed_workload
total_num += elem['cnt']
return compressed_workload, total_num
# parse the explain plan to get estimated cost by database optimizer
@ -425,6 +466,7 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No
found_plan = False
hypo_index = False
is_computed = False
select_sql_pos = []
with open(sql_file, 'w') as file:
if SCHEMA:
file.write('SET current_schema = %s;\n' % SCHEMA)
@ -441,10 +483,14 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No
file.write("set explain_perf_mode = 'normal'; \n")
for ind, query in enumerate(workload):
if 'select ' not in query.statement.lower():
workload[ind].cost_list.append(0)
else:
file.write('EXPLAIN ' + query.statement + ';\n')
select_sql_pos.append(ind)
# record ineffective sql and negative sql for candidate indexes
if is_computed:
record_ineffective_negative_sql(index_config[0], query, ind)
file.write('EXPLAIN ' + query.statement + ';\n')
result = run_shell_sql_cmd(sql_file).split('\n')
if os.path.exists(sql_file):
@ -457,15 +503,18 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No
if 'QUERY PLAN' in line:
found_plan = True
if 'ERROR' in line:
workload.pop(i)
if i >= len(select_sql_pos):
raise ValueError("The size of workload is not correct!")
workload[select_sql_pos[i]].cost_list.append(0)
i += 1
if 'hypopg_create_index' in line:
hypo_index = True
if found_plan and '(cost=' in line:
if i >= len(workload):
if i >= len(select_sql_pos):
raise ValueError("The size of workload is not correct!")
query_cost = parse_explain_plan(line)
query_cost *= workload[i].frequency
workload[i].cost_list.append(query_cost)
query_cost *= workload[select_sql_pos[i]].frequency
workload[select_sql_pos[i]].cost_list.append(query_cost)
total_cost += query_cost
found_plan = False
i += 1
@ -481,8 +530,8 @@ def estimate_workload_cost_file(workload, index_config=None, ori_indexes_name=No
ori_indexes_name.add(ind1.strip().split(' ')[1])
else:
ori_indexes_name.add(ind2)
while i < len(workload):
workload[i].cost_list.append(0)
while i < len(select_sql_pos):
workload[select_sql_pos[i]].cost_list.append(0)
i += 1
if index_config:
run_shell_cmd(['SELECT hypopg_reset_index();'])
@ -602,13 +651,13 @@ def get_indexable_columns(table_index_dict):
return query_indexable_columns
def generate_candidate_indexes(workload, workload_table_name, iterate=False):
def generate_candidate_indexes(workload, workload_table_name):
candidate_indexes = []
index_dict = {}
for k, query in enumerate(workload):
table_index_dict = query_index_advisor(query.statement, workload_table_name)
if iterate:
if 'select ' in query.statement.lower():
table_index_dict = query_index_advisor(query.statement, workload_table_name)
need_check = False
query_indexable_columns = get_indexable_columns(table_index_dict)
valid_index_dict = query_index_check(query.statement, query_indexable_columns)
@ -626,22 +675,20 @@ def generate_candidate_indexes(workload, workload_table_name, iterate=False):
need_check = False
else:
break
else:
valid_index_dict = query_index_check(query.statement, table_index_dict)
# filter duplicate indexes
for table in valid_index_dict.keys():
if table not in index_dict.keys():
index_dict[table] = {}
for columns in valid_index_dict[table]:
if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD:
break
workload[k].valid_index_list.append(IndexItem(table, columns))
if not any(re.match(r'%s' % columns, item) for item in index_dict[table]):
column_sql = {columns: [k]}
index_dict[table].update(column_sql)
elif columns in index_dict[table].keys():
index_dict[table][columns].append(k)
# filter duplicate indexes
for table in valid_index_dict.keys():
if table not in index_dict.keys():
index_dict[table] = {}
for columns in valid_index_dict[table]:
if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD:
break
workload[k].valid_index_list.append(IndexItem(table, columns))
if not any(re.match(r'%s' % columns, item) for item in index_dict[table]):
column_sql = {columns: [k]}
index_dict[table].update(column_sql)
elif columns in index_dict[table].keys():
index_dict[table][columns].append(k)
for table, column_sqls in index_dict.items():
for column, sql in column_sqls.items():
print("table: ", table, "columns: ", column)
@ -787,14 +834,6 @@ def infer_workload_cost(workload, config, atomic_config_total):
min_cost = obj.cost_list[num]
total_cost += min_cost
# compute the cost for updating indexes
if 'insert' in obj.statement.lower() or 'delete' in obj.statement.lower():
for index in config:
index_num = get_index_num(index, atomic_config_total)
if index_num == -1:
raise ValueError("The index isn't found for current query!")
if 0 <= index_num < len(workload[ind].cost_list):
total_cost += obj.cost_list[index_num] - obj.cost_list[0]
# record ineffective sql and negative sql for candidate indexes
if is_computed:
record_ineffective_negative_sql(config[-1], obj, ind)
@ -802,12 +841,12 @@ def infer_workload_cost(workload, config, atomic_config_total):
def simple_index_advisor(input_path, max_index_num):
workload = workload_compression(input_path)
workload, workload_count = workload_compression(input_path)
print_header_boundary(" Generate candidate indexes ")
ori_indexes_name = set()
workload_table_name = dict()
display_info = {'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, True)
display_info = {'workloadCount': workload_count, 'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name)
if len(candidate_indexes) == 0:
print("No candidate indexes generated!")
estimate_workload_cost_file(workload, ori_indexes_name=ori_indexes_name)
@ -868,12 +907,12 @@ def greedy_determine_opt_config(workload, atomic_config_total, candidate_indexes
def complex_index_advisor(input_path):
workload = workload_compression(input_path)
workload, workload_count = workload_compression(input_path)
print_header_boundary(" Generate candidate indexes ")
ori_indexes_name = set()
workload_table_name = dict()
display_info = {'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, True)
display_info = {'workloadCount': workload_count, 'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name)
if len(candidate_indexes) == 0:
print("No candidate indexes generated!")
estimate_workload_cost_file(workload, ori_indexes_name=ori_indexes_name)
@ -957,3 +996,4 @@ def main():
if __name__ == '__main__':
main()

168
src/gausskernel/dbmind/tools/index_advisor/index_advisor_workload_driver.py
View File

@ -105,6 +105,7 @@ class IndexInfo:
self.indexname = indexname
self.columns = columns
self.indexdef = indexdef
self.primary_key = False
self.redundant_obj = []
@ -168,7 +169,10 @@ def filter_low_benefit(pos_list, candidate_indexes, multi_iter_mode, workload):
sql_optimzed += 1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0]
negative_ratio = (index.insert_sql_num + index.delete_sql_num + index.update_sql_num) / \
index.total_sql_num
if sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio:
# filter the candidate indexes that do not meet the conditions of optimization
if sql_optimzed / len(index.positive_pos) < 0.1:
remove_list.append(key)
elif sql_optimzed / len(index.positive_pos) < NEGATIVE_RATIO_THRESHOLD < negative_ratio:
remove_list.append(key)
for item in sorted(remove_list, reverse=True):
candidate_indexes.pop(item)
@ -204,6 +208,9 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total,
sql_info = {'sqlDetails': []}
benefit_types = [index.ineffective_pos, index.positive_pos, index.negative_pos]
for category, benefit_type in enumerate(benefit_types):
sql_count = 0
for item in benefit_type:
sql_count += workload[item].frequency
for ind, pos in enumerate(benefit_type):
sql_detail = {}
sql_template = workload[pos].statement
@ -211,23 +218,27 @@ def display_recommend_result(workload, candidate_indexes, index_cost_total,
sql_template = re.sub(pattern, '?', sql_template)
sql_detail['sqlTemplate'] = sql_template
sql_optimzed = (1 - workload[pos].cost_list[cost_list_pos] / workload[pos].cost_list[0])
sql_detail['sql'] = workload[pos].statement
sql_detail['sqlCount'] = sql_count
if category == 1:
sql_detail['optimized'] = '%.2f' % (sql_optimzed * 100)
sql_optimzed = (workload[pos].cost_list[0] -
workload[pos].cost_list[cost_list_pos]) / \
workload[pos].cost_list[cost_list_pos]
sql_detail['optimized'] = '%.3f' % sql_optimzed
sql_detail['correlationType'] = category
sql_info['sqlDetails'].append(sql_detail)
workload_optimized = 1 - index_cost_total[cost_list_pos] / index_cost_total[0]
sql_info['workloadOptimized'] = '%.2f' % (workload_optimized * 100)
workload_optimized = (1 - index_cost_total[cost_list_pos] / index_cost_total[0]) * 100
sql_info['workloadOptimized'] = '%.2f' % (workload_optimized if workload_optimized > 1 else 1)
sql_info['schemaName'] = SCHEMA
sql_info['tbName'] = table_name
sql_info['columns'] = index.columns
sql_info['statement'] = statement
sql_info['dmlCount'] = round(index.total_sql_num)
sql_info['selectRatio'] = round(index.select_sql_num / index.total_sql_num, 2)
sql_info['insertRatio'] = round(index.insert_sql_num / index.total_sql_num, 2)
sql_info['deleteRatio'] = round(index.delete_sql_num / index.total_sql_num, 2)
sql_info['updateRatio'] = round(index.update_sql_num / index.total_sql_num, 2)
sql_info['selectRatio'] = round(index.select_sql_num * 100 / index.total_sql_num, 2)
sql_info['insertRatio'] = round(index.insert_sql_num * 100 / index.total_sql_num, 2)
sql_info['deleteRatio'] = round(index.delete_sql_num * 100 / index.total_sql_num, 2)
sql_info['updateRatio'] = round(100 - sql_info['selectRatio'] - sql_info['insertRatio']
- sql_info['deleteRatio'], 2)
display_info['recommendIndexes'].append(sql_info)
return display_info
@ -236,13 +247,25 @@ def record_redundant_indexes(cur_table_indexes, redundant_indexes):
cur_table_indexes = sorted(cur_table_indexes,
key=lambda index_obj: len(index_obj.columns.split(',')))
# record redundant indexes
has_restore = []
for pos, index in enumerate(cur_table_indexes[:-1]):
is_redundant = False
for candidate_index in cur_table_indexes[pos + 1:]:
if 'UNIQUE INDEX' in index.indexdef:
# ensure that UNIQUE INDEX will not become redundant compared to normal index
if 'UNIQUE INDEX' not in candidate_index.indexdef:
continue
# ensure redundant index not is pkey
elif index.primary_key:
if re.match(r'%s' % candidate_index.columns, index.columns):
candidate_index.redundant_obj.append(index)
redundant_indexes.append(candidate_index)
has_restore.append(candidate_index)
continue
if re.match(r'%s' % index.columns, candidate_index.columns):
is_redundant = True
index.redundant_obj.append(candidate_index)
if is_redundant:
if is_redundant and index not in has_restore:
redundant_indexes.append(index)
@ -252,14 +275,23 @@ def check_useless_index(tables, db):
if not tables:
return whole_indexes, redundant_indexes
tables_string = ','.join(["'%s'" % table for table in tables[SCHEMA]])
sql = "select tablename, indexname, indexdef from pg_indexes where " \
"schemaname='%s' and tablename in (%s) order by tablename " % (SCHEMA, tables_string)
sql = "SELECT c.relname AS tablename, i.relname AS indexname, " \
"pg_get_indexdef(i.oid) AS indexdef, p.contype AS pkey from " \
"pg_index x JOIN pg_class c ON c.oid = x.indrelid JOIN " \
"pg_class i ON i.oid = x.indexrelid LEFT JOIN pg_namespace n " \
"ON n.oid = c.relnamespace LEFT JOIN pg_constraint p ON i.oid = p.conindid" \
"WHERE (c.relkind = ANY (ARRAY['r'::\"char\", 'm'::\"char\"])) AND " \
"(i.relkind = ANY (ARRAY['i'::\"char\", 'I'::\"char\"])) AND " \
"n.nspname = '%s' AND c.relname in (%s) order by c.relname;" % (SCHEMA, tables_string)
res = db.execute(sql)
if res:
cur_table_indexes = list()
for item in res:
cur_columns = re.search(r'\((.*)\)', item[2]).group(1)
cur_columns = re.search(r'\(([^\(\)]*)\)', item[2]).group(1)
cur_index_obj = IndexInfo(SCHEMA, item[0], item[1], cur_columns, item[2])
if item[3]:
cur_index_obj.primary_key = True
whole_indexes.append(cur_index_obj)
if cur_table_indexes and cur_table_indexes[-1].table != item[0]:
record_redundant_indexes(cur_table_indexes, redundant_indexes)
@ -292,30 +324,39 @@ def check_unused_index_workload(whole_indexes, redundant_indexes, workload_index
indexes_name = set(index.indexname for index in whole_indexes)
unused_index = list(indexes_name.difference(workload_indexes))
remove_list = []
for pos, index in enumerate(redundant_indexes):
is_redundant = False
for redundant_obj in index.redundant_obj:
if redundant_obj.indexname not in unused_index:
is_redundant = True
if not is_redundant:
remove_list.append(pos)
for item in sorted(remove_list, reverse=True):
redundant_indexes.pop(item)
print_header_boundary(" Current workload useless indexes ")
if not unused_index:
print("No useless index!")
detail_info['uselessIndexes'] = []
# useless index
unused_index_columns = dict()
for cur_index in unused_index:
if not re.search('_pkey$', cur_index):
for index in whole_indexes:
if cur_index == index.indexname:
for index in whole_indexes:
if cur_index == index.indexname:
unused_index_columns[cur_index] = index.columns
if 'UNIQUE INDEX' not in index.indexdef:
statement = "DROP INDEX %s;" % index.indexname
print(statement)
useless_index = {"schemaName": index.schema, "tbName": index.table, "type": 1,
"columns": index.columns, "statement": statement}
detail_info['uselessIndexes'].append(useless_index)
print_header_boundary(" Redundant indexes ")
# filter redundant index
for pos, index in enumerate(redundant_indexes):
is_redundant = False
for redundant_obj in index.redundant_obj:
# redundant objects are not in the useless index set or
# equal to the column value in the useless index must be redundant index
index_exist = redundant_obj.indexname not in unused_index_columns.keys() or \
(unused_index_columns.get(redundant_obj.indexname) and
redundant_obj.columns == unused_index_columns[redundant_obj.indexname])
if index_exist:
is_redundant = True
if not is_redundant:
remove_list.append(pos)
for item in sorted(remove_list, reverse=True):
redundant_indexes.pop(item)
if not redundant_indexes:
print("No redundant index!")
# redundant index
@ -375,6 +416,7 @@ def get_workload_template(workload):
def workload_compression(input_path):
total_num = 0
compressed_workload = []
if JSON_TYPE:
with open(input_path, 'r') as file:
@ -387,7 +429,8 @@ def workload_compression(input_path):
for sql in elem['samples']:
compressed_workload.append(QueryItem(sql.strip('\n'),
elem['cnt'] / len(elem['samples'])))
return compressed_workload
total_num += elem['cnt']
return compressed_workload, total_num
# parse the explain plan to get estimated cost by database optimizer
@ -447,17 +490,15 @@ def estimate_workload_cost_file(workload, db, index_config=None, ori_indexes_nam
# record ineffective sql and negative sql for candidate indexes
if is_computed:
record_ineffective_negative_sql(index_config[0], query, ind)
res = db.execute('EXPLAIN ' + query.statement)
if res:
query_cost = parse_explain_plan(res, index_config, ori_indexes_name)
query_cost *= workload[ind].frequency
workload[ind].cost_list.append(query_cost)
total_cost += query_cost
if 'select ' not in query.statement.lower():
workload[ind].cost_list.append(0)
else:
remove_list.append(ind)
for item in sorted(remove_list, reverse=True):
workload.pop(item)
res = db.execute('EXPLAIN ' + query.statement)
if res:
query_cost = parse_explain_plan(res, index_config, ori_indexes_name)
query_cost *= workload[ind].frequency
workload[ind].cost_list.append(query_cost)
total_cost += query_cost
if index_config:
db.execute('SELECT hypopg_reset_index()')
return total_cost
@ -572,13 +613,13 @@ def get_indexable_columns(table_index_dict):
return query_indexable_columns
def generate_candidate_indexes(workload, workload_table_name, db, iterate=False):
def generate_candidate_indexes(workload, workload_table_name, db):
candidate_indexes = []
index_dict = {}
db.init_conn_handle()
for k, query in enumerate(workload):
table_index_dict = query_index_advisor(query.statement, workload_table_name, db)
if iterate:
if 'select ' in query.statement.lower():
table_index_dict = query_index_advisor(query.statement, workload_table_name, db)
need_check = False
query_indexable_columns = get_indexable_columns(table_index_dict)
valid_index_dict = query_index_check(query.statement, query_indexable_columns, db)
@ -596,22 +637,20 @@ def generate_candidate_indexes(workload, workload_table_name, db, iterate=False)
need_check = False
else:
break
else:
valid_index_dict = query_index_check(query.statement, table_index_dict, db)
# filter duplicate indexes
for table in valid_index_dict.keys():
if table not in index_dict.keys():
index_dict[table] = {}
for columns in valid_index_dict[table]:
if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD:
break
workload[k].valid_index_list.append(IndexItem(table, columns))
if not any(re.match(r'%s' % columns, item) for item in index_dict[table]):
column_sql = {columns: [k]}
index_dict[table].update(column_sql)
elif columns in index_dict[table].keys():
index_dict[table][columns].append(k)
# filter duplicate indexes
for table in valid_index_dict.keys():
if table not in index_dict.keys():
index_dict[table] = {}
for columns in valid_index_dict[table]:
if len(workload[k].valid_index_list) >= FULL_ARRANGEMENT_THRESHOLD:
break
workload[k].valid_index_list.append(IndexItem(table, columns))
if not any(re.match(r'%s' % columns, item) for item in index_dict[table]):
column_sql = {columns: [k]}
index_dict[table].update(column_sql)
elif columns in index_dict[table].keys():
index_dict[table][columns].append(k)
for table, column_sqls in index_dict.items():
for column, sql in column_sqls.items():
print("table: ", table, "columns: ", column)
@ -757,14 +796,6 @@ def infer_workload_cost(workload, config, atomic_config_total):
min_cost = obj.cost_list[num]
total_cost += min_cost
# compute the cost for updating indexes
if 'insert' in obj.statement.lower() or 'delete' in obj.statement.lower():
for index in config:
index_num = get_index_num(index, atomic_config_total)
if index_num == -1:
raise ValueError("The index isn't found for current query!")
if 0 <= index_num < len(workload[ind].cost_list):
total_cost += obj.cost_list[index_num] - obj.cost_list[0]
# record ineffective sql and negative sql for candidate indexes
if is_computed:
record_ineffective_negative_sql(config[-1], obj, ind)
@ -772,12 +803,12 @@ def infer_workload_cost(workload, config, atomic_config_total):
def simple_index_advisor(input_path, max_index_num, db):
workload = workload_compression(input_path)
workload, workload_count = workload_compression(input_path)
print_header_boundary(" Generate candidate indexes ")
ori_indexes_name = set()
workload_table_name = dict()
display_info = {'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db, True)
display_info = {'workloadCount': workload_count, 'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db)
db.init_conn_handle()
if len(candidate_indexes) == 0:
print("No candidate indexes generated!")
@ -840,12 +871,12 @@ def greedy_determine_opt_config(workload, atomic_config_total, candidate_indexes
def complex_index_advisor(input_path, db):
workload = workload_compression(input_path)
workload, workload_count = workload_compression(input_path)
print_header_boundary(" Generate candidate indexes ")
ori_indexes_name = set()
workload_table_name = dict()
display_info = {'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db, True)
display_info = {'workloadCount': workload_count, 'recommendIndexes': []}
candidate_indexes = generate_candidate_indexes(workload, workload_table_name, db)
db.init_conn_handle()
if len(candidate_indexes) == 0:
print("No candidate indexes generated!")
@ -932,3 +963,4 @@ def main():
if __name__ == '__main__':
main()

262
src/gausskernel/dbmind/tools/index_advisor/index_server.py
View File

@ -53,15 +53,15 @@ class CreateLogger:
self.level = level
self.log_name = log_name
def create_log(self):
def create_log(self, log_path):
logger = logging.getLogger(self.log_name)
log_path = os.path.join(current_dirname, 'log')
log_path = os.path.join(os.path.dirname(log_path), 'log')
if os.path.exists(log_path):
if os.path.isfile(log_path):
os.remove(log_path)
os.mkdir(log_path)
else:
os.mkdir(log_path)
os.makedirs(log_path)
agent_handler = handlers.RotatingFileHandler(filename=os.path.join(log_path, self.log_name),
maxBytes=1024 * 1024 * 100,
backupCount=5)
@ -74,30 +74,11 @@ class CreateLogger:
class IndexServer:
def __init__(self, pid_file, logger, password):
def __init__(self, pid_file, logger, password, **kwargs):
self.pid_file = pid_file
self.logger = logger
self.app_name = None
self.database = None
self.port = None
self.host = None
self.user = None
self.password = password
self.wl_user = None
self.schema = None
self.max_index_num = 0
self.max_index_storage = 0
self.driver = False
self.index_intervals = 0
self.sql_amount = 0
self.max_generate_log_size = 0
self.statement = False
self.output_sql_file = None
self.pg_log_path = None
self.datanode = None
self.ai_monitor_url = None
self.log_min_duration_statement = None
self.log_statement = None
self._kwargs = kwargs
def check_proc_exist(self, proc_name):
"""
@ -123,14 +104,14 @@ class IndexServer:
proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True)
std, err_msg = proc.communicate()
if proc.returncode != 0:
self.logger.error("Failed to execute command: %s, \nError: %s." % (cmd, str(err_msg)))
self.logger.error("Failed to execute command. Error: %s." % str(err_msg))
return proc.returncode, std.decode()
def save_recommendation_infos(self, recommendation_infos):
headers = {'Content-Type': 'application/json'}
data = json.dumps(recommendation_infos, default=lambda o: o.__dict__, sort_keys=True,
indent=4).encode()
request = Request(url=self.ai_monitor_url, headers=headers,
request = Request(url=self._kwargs['ai_monitor_url'], headers=headers,
data=data)
response = None
@ -151,31 +132,31 @@ class IndexServer:
detail_info_pos = pos + 1
break
detail_info_json = json.loads('\n'.join(index_info[detail_info_pos:]))
detail_info_json['appName'] = self.app_name
detail_info_json['nodeHost'] = self.host
detail_info_json['dbName'] = self.database
detail_info_json['appName'] = self._kwargs.get('app_name')
detail_info_json['nodeHost'] = self._kwargs.get('host')
detail_info_json['dbName'] = self._kwargs.get('database')
return detail_info_json
def execute_index_advisor(self):
self.logger.info('Index advisor task starting.')
try:
cmd = 'echo %s | python3 %s/index_advisor_workload.py %s %s %s -U %s --h %s -W ' \
cmd = 'echo %s | python3 %s/index_advisor_workload.py %s %s %s -U %s -W ' \
'--schema %s --json --multi_iter_mode --show_detail' % (
self.password, current_dirname, self.port, self.database,
self.output_sql_file, self.user, self.host, self.schema)
if self.max_index_storage:
cmd += ' --max_index_storage %s ' % self.max_index_storage
if self.max_index_num:
cmd += ' --max_index_num %s ' % self.max_index_num
if self.driver:
self.password, current_dirname, self._kwargs['port'], self._kwargs['database'],
self._kwargs['output_sql_file'], self._kwargs['user'], self._kwargs['schema'])
if self._kwargs['max_index_storage']:
cmd += ' --max_index_storage %s ' % self._kwargs['max_index_storage']
if self._kwargs['max_index_num']:
cmd += ' --max_index_num %s ' % self._kwargs['max_index_num']
if self._kwargs['driver']:
try:
import psycopg2
cmd = cmd.replace('index_advisor_workload.py',
'index_advisor_workload_driver.py')
except ImportError:
self.logger.warning('Driver import failed, use gsql to connect to the database.')
self.logger.info('Index advisor cmd:%s' % cmd)
if os.path.exists(self.output_sql_file):
self.logger.info('Index advisor cmd:%s' % cmd.split('|')[-1])
if os.path.exists(self._kwargs['output_sql_file']):
_, res = self.execute_cmd(cmd)
detail_info_json = self.convert_output_to_recommendation_infos(res)
@ -188,21 +169,70 @@ class IndexServer:
except Exception as e:
self.logger.error(e)
def extract_log(self, start_time):
extract_log_cmd = 'python3 %s %s %s --start_time "%s"' % \
(os.path.join(current_dirname, 'extract_log.py'),
self._kwargs['pg_log_path'],
self._kwargs['output_sql_file'], start_time)
if self._kwargs['database']:
extract_log_cmd += ' -d %s ' % self._kwargs['database']
if self._kwargs['wl_user']:
extract_log_cmd += ' -U %s ' % self._kwargs['wl_user']
if self._kwargs['sql_amount']:
extract_log_cmd += ' --sql_amount %s ' % self._kwargs['sql_amount']
if self._kwargs['statement']:
extract_log_cmd += ' --statement '
self.logger.info('Extracting log cmd: %s' % extract_log_cmd)
self.execute_cmd(extract_log_cmd)
self.logger.info('The current log extraction is complete.')
def monitor_log_size(self, guc_reset):
self.logger.info('Open GUC params.')
# get original all file size
original_total_size = self.get_directory_size()
self.logger.info('Original total file size: %sM' % (original_total_size / 1024 / 1024))
deviation_size = 0
# open guc
guc_reload = 'gs_guc reload -Z datanode -D {datanode} -c "log_min_duration_statement = 0" && ' \
'gs_guc reload -Z datanode -D {datanode} -c "log_statement= \'all\'"' \
.format(datanode=self._kwargs['datanode'])
self.execute_cmd(guc_reload)
start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(time.time())))
# caculate log size
count = 0
while deviation_size < self._kwargs['max_generate_log_size']:
time.sleep(5)
current_size = self.get_directory_size()
deviation_size = (current_size - original_total_size) / 1024 / 1024
if current_size - original_total_size < 0:
if count >= 60:
break
count += 1
self.logger.info('Current log size difference: %sM' % deviation_size)
self.logger.info('Start to reset GUC, cmd: %s' % guc_reset)
returncode, res = self.execute_cmd(guc_reset)
if returncode == 0:
self.logger.info('Success to reset GUC setting.')
else:
self.logger.error('Failed to reset GUC params. please check it.')
return start_time
def get_directory_size(self):
files = os.listdir(self.pg_log_path)
files = os.listdir(self._kwargs['pg_log_path'])
total_size = 0
for file in files:
total_size += os.path.getsize(os.path.join(self.pg_log_path, file))
total_size += os.path.getsize(os.path.join(self._kwargs['pg_log_path'], file))
return total_size
def execute_log_index_advisor(self):
def execute_index_recommendation(self):
self.logger.info('Start checking guc.')
try:
guc_check = 'gs_guc check -Z datanode -D {datanode} -c "log_min_duration_statement" && ' \
'gs_guc check -Z datanode -D {datanode} -c "log_statement" '.format(datanode=self.datanode)
'gs_guc check -Z datanode -D {datanode} -c "log_statement" '\
.format(datanode=self._kwargs['datanode'])
returncode, res = self.execute_cmd(guc_check)
origin_min_duration = self.log_min_duration_statement
origin_log_statement = self.log_statement
origin_min_duration = self._kwargs['log_min_duration_statement']
origin_log_statement = self._kwargs['log_statement']
if returncode == 0:
self.logger.info('Original GUC settings is: %s' % res)
match_res = re.findall(r'log_min_duration_statement=(\'?[a-zA-Z0-9]+\'?)', res)
@ -220,99 +250,33 @@ class IndexServer:
self.logger.info('Test reseting GUC command...')
guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \
'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \
(self.datanode, origin_min_duration, self.datanode, origin_log_statement)
(self._kwargs['datanode'], origin_min_duration,
self._kwargs['datanode'], origin_log_statement)
returncode, res = self.execute_cmd(guc_reset)
if returncode != 0:
guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \
'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \
(self.datanode, self.log_min_duration_statement, self.datanode, self.log_statement)
(self._kwargs['datanode'], self._kwargs['log_min_duration_statement'],
self._kwargs['datanode'], self._kwargs['log_statement'])
ret, res = self.execute_cmd(guc_reset)
if ret != 0:
raise Exception('Cannot reset GUC initial value, please check it.')
self.logger.info('Test successfully')
self.logger.info('Open GUC params.')
# get original all file size
original_total_size = self.get_directory_size()
self.logger.info('Original total file size: %sM' % (original_total_size/1024/1024))
deviation_size = 0
# open guc
guc_reload = 'gs_guc reload -Z datanode -D {datanode} -c "log_min_duration_statement = 0" && ' \
'gs_guc reload -Z datanode -D {datanode} -c "log_statement= \'all\'"'.format(datanode=self.datanode)
self.execute_cmd(guc_reload)
start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(time.time())))
# caculate log size
count = 0
while deviation_size < self.max_generate_log_size:
time.sleep(5)
current_size = self.get_directory_size()
deviation_size = (current_size - original_total_size)/1024/1024
if current_size - original_total_size < 0:
if count >= 60:
break
count += 1
self.logger.info('Current log size difference: %sM' % deviation_size)
self.logger.info('Start to reset GUC, cmd: %s' % guc_reset)
returncode, res = self.execute_cmd(guc_reset)
if returncode == 0:
self.logger.info('Success to reset GUC setting.')
else:
self.logger.error('Failed to reset GUC params. please check it.')
# open guc and monitor log real-time size
start_time = self.monitor_log_size(guc_reset)
# extract log
extract_log_cmd = 'python3 %s %s %s -d %s -U %s --start_time "%s"' % \
(os.path.join(current_dirname, 'extract_log.py'),
self.pg_log_path, self.output_sql_file,
self.database, self.wl_user, start_time)
if self.sql_amount:
extract_log_cmd += ' --sql_amount %s ' % self.sql_amount
if self.statement:
extract_log_cmd += ' --statement '
self.logger.info('Extracting log cmd: %s' % extract_log_cmd)
self.execute_cmd(extract_log_cmd)
self.logger.info('The current log extraction is complete.')
# index adviss
self.extract_log(start_time)
# index advise
self.execute_index_advisor()
except Exception as e:
self.logger.error(e)
guc_reset = 'gs_guc reload -Z datanode -D %s -c "log_min_duration_statement = %s" && ' \
'gs_guc reload -Z datanode -D %s -c "log_statement= %s"' % \
(self.datanode, self.log_min_duration_statement, self.datanode,
self.log_statement)
(self._kwargs['datanode'], self._kwargs['log_min_duration_statement'],
self._kwargs['datanode'], self._kwargs['log_statement'])
self.execute_cmd(guc_reset)
def parse_check_conf(self, config_path):
config = ConfigParser()
config.read(config_path)
self.app_name = config.get("server", "app_name")
self.database = config.get("server", "database")
self.port = config.get("server", "port")
self.host = config.get("server", "host")
self.user = config.get("server", "user")
self.wl_user = config.get("server", "workload_user")
self.schema = config.get("server", "schema")
self.max_index_num = config.getint("server", "max_index_num")
self.max_index_storage = config.get("server", "max_index_storage")
self.driver = config.getboolean("server", "driver")
self.index_intervals = config.getint("server", "index_intervals")
self.sql_amount = config.getint("server", "sql_amount")
self.output_sql_file = config.get("server", "output_sql_file")
self.datanode = config.get("server", "datanode")
self.pg_log_path = config.get("server", "pg_log_path")
self.ai_monitor_url = config.get("server", "ai_monitor_url")
self.max_generate_log_size = config.getfloat("server", "max_generate_log_size")
self.statement = config.getboolean("server", "statement")
self.log_min_duration_statement = config.get("server", "log_min_duration_statement")
self.log_statement = config.get("server", "log_statement")
if not self.log_min_duration_statement or not re.match(r'[a-zA-Z0-9]+',
self.log_min_duration_statement):
raise ValueError("Please enter a legal value of [log_min_duration_statement]")
legal_log_statement = ['none', 'all', 'ddl', 'mod']
if self.log_statement not in legal_log_statement:
raise ValueError("Please enter a legal value of [log_statement]")
def start_service(self, config_path):
def start_service(self):
# check service is running or not.
if os.path.isfile(self.pid_file):
pid = self.check_proc_exist("index_server")
@ -320,9 +284,6 @@ class IndexServer:
raise Exception("Error: Process already running, can't start again.")
else:
os.remove(self.pid_file)
# check config file exists
if not os.path.isfile(config_path):
raise Exception("Config file: %s does not exists." % config_path)
# get listen host and port
self.logger.info("Start service...")
@ -332,12 +293,12 @@ class IndexServer:
with open(self.pid_file, mode='w') as f:
f.write(str(os.getpid()))
self.parse_check_conf(config_path)
self.logger.info("Index advisor execution intervals is: %sh" % self.index_intervals)
index_advisor_thread = RepeatTimer(self.index_intervals*60*60, self.execute_log_index_advisor)
self.logger.info("Index advisor execution intervals is: %sh" %
self._kwargs['index_intervals'])
index_recommendation_thread = RepeatTimer(self._kwargs['index_intervals']*60*60,
self.execute_index_recommendation)
self.logger.info("Start timer...")
index_advisor_thread.start()
index_recommendation_thread.start()
def read_input_from_pipe():
@ -356,12 +317,46 @@ def read_input_from_pipe():
return input_str
def parse_check_conf(config_path):
config = ConfigParser()
config.read(config_path)
config_dict = dict()
config_dict['app_name'] = config.get("server", "app_name")
config_dict['database'] = config.get("server", "database")
config_dict['port'] = config.get("server", "port")
config_dict['host'] = config.get("server", "host")
config_dict['user'] = config.get("server", "user")
config_dict['wl_user'] = config.get("server", "workload_user")
config_dict['schema'] = config.get("server", "schema")
config_dict['max_index_num'] = config.getint("server", "max_index_num")
config_dict['max_index_storage'] = config.get("server", "max_index_storage")
config_dict['driver'] = config.getboolean("server", "driver")
config_dict['index_intervals'] = config.getint("server", "index_intervals")
config_dict['sql_amount'] = config.getint("server", "sql_amount")
config_dict['output_sql_file'] = config.get("server", "output_sql_file")
config_dict['datanode'] = config.get("server", "datanode")
config_dict['pg_log_path'] = config.get("server", "pg_log_path")
config_dict['ai_monitor_url'] = config.get("server", "ai_monitor_url")
config_dict['max_generate_log_size'] = config.getfloat("server", "max_generate_log_size")
config_dict['statement'] = config.getboolean("server", "statement")
config_dict['log_min_duration_statement'] = config.get("server", "log_min_duration_statement")
config_dict['log_statement'] = config.get("server", "log_statement")
if not config_dict['log_min_duration_statement'] or \
not re.match(r'[a-zA-Z0-9]+', config_dict['log_min_duration_statement']):
raise ValueError("Please enter a legal value of [log_min_duration_statement]")
legal_log_statement = ['none', 'all', 'ddl', 'mod']
if config_dict['log_statement'] not in legal_log_statement:
raise ValueError("Please enter a legal value of [log_statement]")
return config_dict
def manage_service():
LOGGER = CreateLogger("debug", "start_service.log").create_log()
config_path = os.path.join(current_dirname, 'database-info.conf')
config_dict = parse_check_conf(config_path)
LOGGER = CreateLogger("debug", "start_service.log").create_log(config_dict.get('output_sql_file'))
server_pid_file = os.path.join(current_dirname, 'index_server.pid')
password = read_input_from_pipe()
IndexServer(server_pid_file, LOGGER, password).start_service(
os.path.join(current_dirname, 'database-info.conf'))
IndexServer(server_pid_file, LOGGER, password, **config_dict).start_service()
def main():
@ -375,3 +370,4 @@ def main():
if __name__ == '__main__':
main()