import random
import time
import numpy as np
import pandas as pd
from db.conf import table_suffix
"""
A tool class that provides QDG algorithm and computes hyper / shuffle join cost.
"""
class JoinEvaluator:
def __init__(self,join_queryset, joined_cols,joined_trees,joined_tables,block_size,metadata,benchmark='tpch'):
self.A_join_queries=[join_queryset[0]]
self.B_join_queries=[join_queryset[1]]
self.join_tables=joined_tables
self.pa_A=joined_trees[0]
self.pa_B=joined_trees[1]
self.join_attr=joined_cols
self.metadata=metadata
self.benchmark=benchmark
self.dim_nums1,self.dim_nums2=len(join_queryset[0])//2,len(join_queryset[1])//2
self.worker_mermory=10
def generate_join_queries(self,a_training_set_for_join,b_training_set_for_join,join_amount=20):
join_attr=self.join_attr
def __overlap(q1, q2, dim1, dim2):
if q1[dim] <= q2[dim] <= q1[dim + self.dim_nums1] or q2[dim] <= q1[dim] <= q2[dim + self.dim_nums2]:
return True
return False
a_training_set=a_training_set_for_join
b_training_set=b_training_set_for_join
# pick join query which will be measure
b_join_index = []
for _ in range(join_amount):
b_join_index.append(
list(set([random.randint(0, len(b_training_set) - 1) for _ in range(random.randint(1, 10))])))
# remove block id with overlap join attribute range
b_join_queries = []
for ids in b_join_index:
item = []
for idx in ids:
flag = True
for em in item:
if __overlap(b_training_set[idx], em, join_attr):
flag = False
break
if flag: item.append(b_training_set[idx])
b_join_queries.append(item)
a_join_queries = {}
for bid, item in enumerate(b_join_queries):
for qb in item:
a_join_queries[bid] = []
for qa in a_training_set:
if __overlap(qa, qb, join_attr):
# remove overlap range queries
flag = True
for qa2 in a_join_queries[bid]:
# if __overlap(qa2,qa,join_attr):
if qa2 == qa:
flag = False
break
if flag: a_join_queries[bid].append(qa)
return a_join_queries,b_join_queries
# def rough_join_cost(self,group_type):
# hyper_read_cost, hyper_read_bytes, temp_joined_df, group_time= self.compute_total_shuffle_hyper_cost(group_type)
# # if shuffle:
# # return shuffle_read_cost
# return hyper_read_cost,hyper_read_bytes,temp_joined_df,group_time
def pandas_hash_join(self,dataset_A, dataset_B, key_A, key_B):
A_used_columns=self.pa_A.used_columns
B_used_columns=self.pa_B.used_columns
if self.benchmark=='imdb':
A_used_columns=[table_suffix[self.benchmark][self.join_tables[0]]+'_'+col for col in A_used_columns]
B_used_columns=[table_suffix[self.benchmark][self.join_tables[1]]+'_'+col for col in B_used_columns]
df_A = pd.DataFrame(dataset_A, columns=A_used_columns)
df_B = pd.DataFrame(dataset_B, columns=B_used_columns)
left_col = A_used_columns[key_A]
right_col = B_used_columns[key_B]
if df_A.shape[0]>df_B.shape[0]:
df_A,df_B=df_B,df_A
left_col,right_col=right_col,left_col
df_B = df_B.drop_duplicates(subset=[right_col], keep='first')
merged_df = df_A.merge(df_B, how='inner', left_on=left_col, right_on=right_col)
return merged_df
def compute_total_shuffle_hyper_cost(self,group_type,is_real_hyper):
start_time = time.time()
pa_A=self.pa_A
pa_B=self.pa_B
key_A,key_B=self.join_attr[0],self.join_attr[1]
blocks_A_ids = {}
blocks_B_ids = {}
for qid, query in enumerate(self.A_join_queries):
blocks_A_ids[qid]=list(set(pa_A.query_single(query)))
for qid, query in enumerate(self.B_join_queries):
blocks_B_ids[qid]=list(set(pa_B.query_single(query)))
print(f"Table {self.join_tables[0]} has {len(blocks_A_ids[0])} Blocks, Table {self.join_tables[1]} has {len(blocks_B_ids[0])} Blocks.")
print(f"Read Cost: {sum([pa_A.nid_node_dict[a_id].node_size for a_id in blocks_A_ids[0]])} -> {sum([pa_B.nid_node_dict[b_id].node_size for b_id in blocks_B_ids[0]])}")
static_A_blocks=blocks_A_ids[0].copy()
static_B_blocks=blocks_B_ids[0].copy()
def is_overlay(aid, bid):
bucket_a = pa_A.nid_node_dict[aid].boundary
bucket_b = pa_B.nid_node_dict[bid].boundary
return __overlap(bucket_a, bucket_b, key_A,key_B)
def __overlap(q1, q2, dim1, dim2):
if q1[dim1] <= q2[dim2] <= q1[dim1 + self.dim_nums1] or q2[dim2] <= q1[dim1] <= q2[dim2 + self.dim_nums2]:
return True
return False
final_resized_splits = []
overlap_chunks_for_queries = []
build_time = 0
for qid in range(len(blocks_A_ids)):
A_join_block_ids=blocks_A_ids[qid]
B_join_block_ids=blocks_B_ids[qid]
# group algorithm
# step1: generate overlap_chunks
overlap_chunks = {}
for A_bid in A_join_block_ids:
if A_bid not in overlap_chunks.keys():
overlap_chunks[A_bid] = []
for B_bid in B_join_block_ids:
if is_overlay(A_bid, B_bid):
overlap_chunks[A_bid].append(B_bid)
overlap_chunks_for_queries.append(overlap_chunks)
# step2: group
time0 = time.time()
if group_type==0:
resizedSplits = self.group(overlap_chunks, A_join_block_ids, min_partition_size=self.worker_mermory)
elif group_type==1:
resizedSplits = self.group1(overlap_chunks, A_join_block_ids, min_partition_size=self.worker_mermory,max_partition_size=1.2*self.worker_mermory)
build_time += time.time() - time0
final_resized_splits.append(resizedSplits)
group_time = time.time() - start_time
total_shuffle_hyper_read_cost = 0
a_hyper_cost,b_hyper_cost=0,0
a_shuffle_cost,b_shuffle_cost=0,0
shuffle_weight = 3
for q_no, resizedSplits in enumerate(final_resized_splits):
total_A_ids = []
total_B_ids = []
overlap_chunks = overlap_chunks_for_queries[q_no]
group_a_cost,group_b_cost=0,0
for group in resizedSplits:
B_ids = []
for a_id in group:
group_a_cost+=pa_A.nid_node_dict[a_id].node_size
B_ids += overlap_chunks[a_id]
total_A_ids+=group
B_ids = list(set(B_ids))
total_B_ids += B_ids
for b_id in B_ids:
group_b_cost+=pa_B.nid_node_dict[b_id].node_size
a_hyper_cost+=group_a_cost
b_hyper_cost+=group_b_cost
total_B_ids = list(set(total_B_ids))
for a_id in total_A_ids:
a_shuffle_cost += shuffle_weight * pa_A.nid_node_dict[a_id].node_size
for b_id in total_B_ids:
b_shuffle_cost += shuffle_weight * pa_B.nid_node_dict[b_id].node_size
dataset_A,dataset_B=[],[]
for a_id in static_A_blocks:
dataset_A.extend(pa_A.nid_node_dict[a_id].dataset)
for b_id in static_B_blocks:
dataset_B.extend(pa_B.nid_node_dict[b_id].dataset)
joined_df=self.pandas_hash_join(np.array(dataset_A),np.array(dataset_B),key_A,key_B)
if is_real_hyper:
total_shuffle_hyper_read_cost=a_hyper_cost+b_hyper_cost
total_hyper_shuffle_read_bytes=a_hyper_cost*self.metadata[self.join_tables[0]]['read_line']+b_hyper_cost*self.metadata[self.join_tables[1]]['read_line']
else:
total_shuffle_hyper_read_cost=a_shuffle_cost+b_shuffle_cost
total_hyper_shuffle_read_bytes=a_shuffle_cost*self.metadata[self.join_tables[0]]['read_line']+b_shuffle_cost*self.metadata[self.join_tables[1]]['read_line']
print(f"Hyper Join: {a_hyper_cost} -> {b_hyper_cost}")
print(f"Real Join Type: {is_real_hyper}, Final Cost: {total_shuffle_hyper_read_cost}")
return total_shuffle_hyper_read_cost,total_hyper_shuffle_read_bytes,joined_df,group_time
def print_shuffle_hyper_blocks(self,a_join_queries,b_join_queries,group_type):
pa_A=self.pa_A
pa_B=self.pa_B
join_attr=self.join_attr
blocks_a_ids = []
blocks_b_ids = []
a_join_info = []
b_join_info = []
# how to get join attr range base on block id.
for key, queries in enumerate(b_join_queries):
map_content = {}
join_keys = []
block_ids = []
for query in queries:
join_keys += pa_B.query_single_toplayer(query)
block_ids += pa_B.query_single(query)
map_content[key] = list(set(block_ids))
blocks_b_ids.append(map_content)
join_keys = list(set(join_keys))
join_info = {"nums": len(join_keys), "length": []}
for join_id in join_keys:
node = pa_B.nid_node_dict[join_id]
join_info["length"].append(node.boundary[join_attr + node.num_dims] - node.boundary[join_attr])
b_join_info.append(join_info)
for key in a_join_queries:
map_content = {}
join_keys = []
block_ids = []
for query in a_join_queries[key]:
join_keys += pa_A.query_single_toplayer(query)
block_ids += pa_A.query_single(query)
map_content[key] = list(set(block_ids))
blocks_a_ids.append(map_content)
join_keys = list(set(join_keys))
join_info = {"nums": len(join_keys), "length": []}
for join_id in join_keys:
node = pa_A.nid_node_dict[join_id]
join_info["length"].append(node.boundary[join_attr + node.num_dims] - node.boundary[join_attr])
a_join_info.append(join_info)
join_infos = [a_join_info, b_join_info]
for join_info in join_infos:
total_nums, total_length = 0, 0
for item in join_info:
total_nums += item['nums']
total_length += sum(item['length'])
# print(f"{total_nums} , {round(total_length, 2)}")
# compute hyper join cost （Use Group 4）
def is_overlay(aid, bid):
bucket_a = pa_A.nid_node_dict[aid].boundary
bucket_b = pa_B.nid_node_dict[bid].boundary
return __overlap(bucket_a, bucket_b, join_attr)
def __overlap(q1, q2, dim):
if q1[dim] <= q2[dim] <= q1[dim + self.dim_nums1] or q2[dim] <= q1[dim] <= q2[dim + self.dim_nums2]:
return True
return False
final_resized_splits = []
overlap_chunks_for_queries = []
intersection_reward = 0
total_hyper_cost = 0
build_time = 0
for idx in range(len(blocks_a_ids)):
# if idx<=2:continue
A_join_block_ids = []
for key in blocks_a_ids[idx].keys():
A_join_block_ids += blocks_a_ids[idx][key]
B_join_block_ids = []
for key in blocks_b_ids[idx].keys():
B_join_block_ids += blocks_b_ids[idx][key]
# group algorithm
# step1: generate overlap_chunks
overlap_chunks = {}
for aid in A_join_block_ids:
if aid not in overlap_chunks.keys(): overlap_chunks[aid] = []
for bid in B_join_block_ids:
if is_overlay(aid, bid): overlap_chunks[aid].append(bid)
# print(f"overlap chunks: ",overlap_chunks)
overlap_chunks_for_queries.append(overlap_chunks)
# step2: group
# print(overlap_chunks)
# print(A_join_block_ids)
time0 = time.time()
if group_type==3:
resizedSplits = self.group3(overlap_chunks, A_join_block_ids, partition_size=8)
elif group_type==1:
resizedSplits = self.group1(overlap_chunks, A_join_block_ids, partition_size=8)
build_time += time.time() - time0
for group in resizedSplits:
all_b_ids = []
for a_id in group:
all_b_ids += overlap_chunks[a_id]
# print(overlap_chunks[a_id])
actual_b_ids = list(set(all_b_ids))
intersection_reward += len(all_b_ids) - len(actual_b_ids)
total_hyper_cost += sum([pa_B.nid_node_dict[_].node_size for _ in actual_b_ids])
final_resized_splits.append(resizedSplits)
# print("total_hyper_cost: ", total_hyper_cost)
# print("average build time: ", build_time / len(blocks_a_ids))
A_ids_for_q, B_ids_for_q = [], []
for q_no, resizedSplits in enumerate(final_resized_splits):
cnt = 0
total_B_ids = []
total_A_ids = []
overlap_chunks = overlap_chunks_for_queries[q_no]
for group in resizedSplits:
b_ids = []
for a_id in group:
b_ids += overlap_chunks[a_id]
cnt += 1
total_A_ids += group
b_ids = list(set(b_ids))
if b_ids:
total_B_ids.append(b_ids)
A_ids_for_q.append(total_A_ids)
B_ids_for_q.append(total_B_ids)
return A_ids_for_q, B_ids_for_q
def compute_join_blocks_for_main_table(self,a_join_queries,b_join_queries):
pa_A=self.pa_A
pa_B=self.pa_B
join_attr=self.join_attr
blocks_a_ids = []
blocks_b_ids = []
a_join_info = []
b_join_info = []
# how to get join attr range base on block id.
for key, queries in enumerate(b_join_queries):
map_content = {}
join_keys = []
block_ids = []
for query in queries:
join_keys += pa_B.query_single_toplayer(query)
block_ids += pa_B.query_single(query)
map_content[key] = list(set(block_ids))
blocks_b_ids.append(map_content)
join_keys = list(set(join_keys))
join_info = {"nums": len(join_keys), "length": []}
for join_id in join_keys:
node = pa_B.nid_node_dict[join_id]
join_info["length"].append(node.boundary[join_attr + node.num_dims] - node.boundary[join_attr])
b_join_info.append(join_info)
for key in a_join_queries:
map_content = {}
join_keys = []
block_ids = []
for query in a_join_queries[key]:
join_keys += pa_A.query_single_toplayer(query)
block_ids += pa_A.query_single(query)
map_content[key] = list(set(block_ids))
blocks_a_ids.append(map_content)
join_keys = list(set(join_keys))
join_info = {"nums": len(join_keys), "length": []}
for join_id in join_keys:
node = pa_A.nid_node_dict[join_id]
join_info["length"].append(node.boundary[join_attr + node.num_dims] - node.boundary[join_attr])
a_join_info.append(join_info)
join_infos = [a_join_info, b_join_info]
for join_info in join_infos:
total_nums, total_length = 0, 0
for item in join_info:
total_nums += item['nums']
total_length += sum(item['length'])
a_hyper_blocks_size=0
b_hyper_blocks_size=0
for idx in range(len(blocks_a_ids)):
# if idx<=2:continue
for key in blocks_a_ids[idx].keys():
a_hyper_blocks_size+=sum([pa_A.nid_node_dict[ida].node_size for ida in blocks_a_ids[idx][key]])
for key in blocks_b_ids[idx].keys():
b_hyper_blocks_size+=sum([pa_B.nid_node_dict[idb].node_size for idb in blocks_b_ids[idx][key]])
return a_hyper_blocks_size,0
# (AdaptDB grouping)
def group(self,overlap_chunks,join_a_block_ids, min_partition_size):
def get_intersection_size_count(setValues, listValues):
size = 0
for lv in listValues:
if lv in setValues: size += 1
return size
def get_group_size(a_ids):
chunks = []
for a_id in a_ids:
chunks += overlap_chunks[a_id]
chunks = list(set(chunks))
return sum([self.pa_B.nid_node_dict[b_id].node_size for b_id in chunks])+sum([self.pa_A.nid_node_dict[a_id].node_size for a_id in a_ids])
def __get_group_size(a_ids):
# return sum([self.pa_A.nid_node_dict[a_id].node_size for a_id in a_ids])
return len(a_ids)
resizedSplits = []
rest_count = len(join_a_block_ids)
while rest_count > 0:
cur_splits = []
chunks = []
available_size=min_partition_size
# max block size limit for every split.
while rest_count > 0:
maxIntersection = -1
best_offset = -1
for offset, bid in enumerate(join_a_block_ids):
cur_intersection = get_intersection_size_count(chunks, overlap_chunks[bid])
if cur_intersection > maxIntersection:
maxIntersection = cur_intersection
best_offset = offset
bucket_id = join_a_block_ids[best_offset]
# if _get_group_size(cur_splits+[bucket_id]) > available_size:
# if len(cur_splits) == 0:
# cur_splits.append(bucket_id)
# chunks += overlap_chunks[bucket_id]
# chunks = list(set(chunks))
# join_a_block_ids.remove(bucket_id)
# rest_count -= 1
# break
# break
cur_splits.append(bucket_id)
chunks += overlap_chunks[bucket_id]
chunks = list(set(chunks))
join_a_block_ids.remove(bucket_id)
rest_count -= 1
if __get_group_size(cur_splits) == available_size:
break
resizedSplits.append(cur_splits)
return resizedSplits
# Our QDG grouping algorithm
def group1(self, overlap_chunks, join_a_block_ids, min_partition_size, max_partition_size):
def list_solved_list(l1, l2):
for item1 in l1:
if item1 in l2:
return True
return False
def get_group_size(a_ids):
chunks = []
for a_id in a_ids:
chunks += overlap_chunks[a_id]
chunks = list(set(chunks))
return sum([self.pa_B.nid_node_dict[b_id].node_size for b_id in chunks])
def __get_group_size(a_ids):
# return sum([self.pa_A.nid_node_dict[a_id].node_size for a_id in a_ids])
return len(a_ids)
def get_intersection_size_count(setValues, listValues):
size = 0
for lv in listValues:
if lv in setValues: size += 1
return size
resizedSplits = []
rest_count = len(join_a_block_ids)
affinity_tab = []
pre_save_ids = []
computed_ids_dict = {}
for bid in join_a_block_ids: computed_ids_dict[bid] = {}
a_block_len = len(join_a_block_ids)
for no1 in range(a_block_len):
bid1 = join_a_block_ids[no1]
max_intersection = -1
max_bid = []
for exist_bid in computed_ids_dict[bid1].keys():
cur_intersection = computed_ids_dict[bid1][exist_bid]
if cur_intersection > max_intersection:
max_intersection = cur_intersection
max_bid = [exist_bid]
for no2 in range(no1 + 1, a_block_len):
bid2 = join_a_block_ids[no2]
cur_intersection = get_intersection_size_count(overlap_chunks[bid1], overlap_chunks[bid2])
computed_ids_dict[bid1][bid2] = cur_intersection
computed_ids_dict[bid2][bid1] = cur_intersection
if cur_intersection > max_intersection:
max_intersection = cur_intersection
max_bid = [bid2]
if max_intersection == 0:
pre_save_ids.append(bid1)
else:
affinity_tab.append({'item': [[bid1], max_bid], 'val': max_intersection, 'chunk': overlap_chunks[bid1]})
cur_index = 0
# pre-save these ids which doesn't have any overlap blocks
last_index=0
while last_index < len(pre_save_ids):
if __get_group_size(pre_save_ids[last_index:cur_index + 1]) == min_partition_size:
merge_ids = pre_save_ids[last_index:cur_index + 1]
resizedSplits.append(merge_ids)
rest_count -= len(merge_ids)
last_index = cur_index + 1
elif cur_index >= len(pre_save_ids)-1:
merge_ids = pre_save_ids[last_index:]
resizedSplits.append(merge_ids)
rest_count -= len(merge_ids)
break
else:
cur_index += 1
while rest_count > 0:
affinity_tab.sort(key=lambda item: (item['val'], len(item['item'][0])), reverse=True)
sel_tab = affinity_tab.pop(0)
merge_ids = sel_tab['item'][0] + sel_tab['item'][1]
merge_ids_length = len(merge_ids)
is_completed = False
if __get_group_size==min_partition_size or len(affinity_tab)==0 or sel_tab['val']==-1:
is_completed = True
resizedSplits.append(merge_ids)
rest_count -= merge_ids_length
else:
# add key=chunk
new_overlap_chunks = sel_tab['chunk']
for bid in sel_tab['item'][1]:
new_overlap_chunks += overlap_chunks[bid]
new_tab = {'item': [merge_ids, []], 'val': -1, 'chunk': list(set(new_overlap_chunks))}
# update affinity_tab
for tab in reversed(affinity_tab):
# delete tab
if list_solved_list(tab['item'][0], sel_tab['item'][1]):
affinity_tab.remove(tab)
continue
# update tab
if list_solved_list(tab['item'][1], merge_ids):
if is_completed or __get_group_size(tab['item'][0]) + __get_group_size(merge_ids) > min_partition_size:
tab['item'][1] = []
tab['val'] = -1
else:
tab['item'][1] = merge_ids
tab['val'] = get_intersection_size_count(tab['chunk'], new_tab['chunk'])
if not is_completed: affinity_tab.append(new_tab)
# Case: the affinity_tab only has one item.
if len(affinity_tab) == 1:
last_tab = affinity_tab.pop(0)
merge_ids = last_tab['item'][0] + last_tab['item'][1]
resizedSplits.append(merge_ids)
rest_count -= len(merge_ids)
# Update overlap_chunks and intersection benefits for merged column groups
for ud_item1 in affinity_tab:
if ud_item1['val'] == -1:
ud1_key = ud_item1['item'][0]
flag1 = False
if len(ud1_key) == 1: flag1 = True
overlap_chunks1 = ud_item1['chunk']
max_allocate_size = min_partition_size - __get_group_size(ud1_key)
max_intersection = -1
max_target_ids = []
for ud_item2 in affinity_tab:
ud2_key = ud_item2['item'][0]
if ud1_key == ud2_key: continue
if __get_group_size(ud2_key) > max_allocate_size: continue
if ud_item2['item'][1] == ud1_key:
cur_intersection = ud_item2['val']
else:
# if flag1 and len(ud2_key)==1:continue
flag2 = False
if len(ud2_key) == 1: flag2 = True
if flag1 and flag2:
cur_intersection = computed_ids_dict[ud1_key[0]][ud2_key[0]]
else:
overlap_chunks2 = ud_item2['chunk']
cur_intersection = get_intersection_size_count(overlap_chunks1, overlap_chunks2)
if cur_intersection > max_intersection:
max_intersection = cur_intersection
max_target_ids = ud2_key
ud_item1['val'] = max_intersection
ud_item1['item'][1] = max_target_ids
return resizedSplits