Function sub_process

union/D3L/query.py:34–73

(query_tables, queue, output_folder, qe, dataloader, k_value)

    return result

def sub_process(query_tables, queue, output_folder, qe, dataloader, k_value):
    for i, table_name_with_extension in enumerate(query_tables):

        table_name = os.path.splitext(table_name_with_extension)[0]
        if not os.path.exists(output_folder):
            os.makedirs(output_folder)

        results_file = os.path.join(output_folder, f"{table_name}.csv")

        if os.path.exists(results_file):
            print(f"跳过查询表 {i + 1}，因为结果文件已经存在：{results_file}")
            queue.put(1)  # 在队列中放入一个占位符值
            continue

        # 执行查询
        results, extended_results = qe.table_query(table=dataloader.read_table(table_name=table_name),
                                                   aggregator=None, k=k_value, verbose=True)

        # 创建一个新的 CSV 文件
        with open(results_file, mode='w', newline='') as csvfile:
            writer = csv.writer(csvfile)

            # Write the header
            writer.writerow(['query_table', 'candidate_table', 'query_col_name', 'candidate_col_name'])

            # 写入查询结果到 CSV 文件
            for result in extended_results:
                query_table = table_name
                candidate_table = os.path.basename(result[0])

                for x, column_info in enumerate(result[1]):
                    column_info_name = f"column_info{x+1}"
                    globals()[column_info_name] = column_info[0]

                    row = [f"{query_table}.csv", f"{candidate_table}.csv", column_info[0][0], column_info[0][1]]
                    writer.writerow(row)

        print(f"Results for query table {i + 1} have been written to {results_file}")
        queue.put(1)
    queue.put((-1, "test-pid"))

def merge_csv(output_folder, combined_file_path):
    csv_files = glob.glob(output_folder + "/*.csv")