From 0300e1e89cfb490071d0afd0ba75804393ad0c32 Mon Sep 17 00:00:00 2001 From: acanadas Date: Tue, 17 Jun 2025 14:19:08 +0000 Subject: [PATCH 1/5] adding ROCpd database merge --- source/lib/python/rocpd/__main__.py | 38 +++ source/lib/python/rocpd/merge.py | 363 ++++++++++++++++++++++++++++ source/lib/python/utilities.cmake | 1 + 3 files changed, 402 insertions(+) create mode 100644 source/lib/python/rocpd/merge.py diff --git a/source/lib/python/rocpd/__main__.py b/source/lib/python/rocpd/__main__.py index 92f5eaf4e9..627d78a41d 100644 --- a/source/lib/python/rocpd/__main__.py +++ b/source/lib/python/rocpd/__main__.py @@ -39,6 +39,7 @@ def main(argv=None, config=None): """ import argparse from . import csv + from . import merge from . import otf2 from . import output_config from . import pftrace @@ -64,6 +65,13 @@ def main(argv=None, config=None): Convert 2 databases, output CSV, OTF2, and perfetto trace formats $ rocpd convert -i db{3,4}.db --output-format csv otf2 pftrace +""" + + merge_examples = """ + +Example usage: + + TODO: Add examples for merge command """ query_examples = """ @@ -118,6 +126,14 @@ def main(argv=None, config=None): epilog=convert_examples, ) + merger = subparsers.add_parser( + "merge", + description="Generate merged database from rocPD databases", + allow_abbrev=False, + formatter_class=argparse.RawTextHelpFormatter, + epilog=merge_examples, + ) + query_reporter = subparsers.add_parser( "query", description="Generate output on a query", @@ -158,6 +174,16 @@ def get_output_type(val): required=True, ) + merger_required_params = merger.add_argument_group("Required options") + merger_required_params.add_argument( + "-i", + "--input", + required=True, + type=output_config.check_file_exists, + nargs="+", + help="Input path and filename to one or more database(s)", + ) + query_required_params = query_reporter.add_argument_group("Required options") query_required_params.add_argument( "-i", @@ -186,6 +212,9 @@ def get_output_type(val): valid_otf2_args = otf2.add_args(converter) valid_time_window_args = time_window.add_args(converter) + # merge: subparser args + valid_merge_args = merge.add_args(merger) + # query: subparser args valid_out_config_args = output_config.add_args(query_reporter) valid_query_args = query.add_args(query_reporter) @@ -258,6 +287,15 @@ def get_output_type(val): else: print(f"Warning: Unsupported output format '{out_format}'") + # if the user requested merge module, execute the merge + elif args.command == "merge": + # merge subparser args + merge_args = merge.process_args(args, valid_merge_args) + + # now start processing the data. Import the data and merge the views + importData = RocpdImportData(args.input) + merge.merge(importData, **merge_args) + # if the user requested query module, execute the query elif args.command == "query": # query subparser args diff --git a/source/lib/python/rocpd/merge.py b/source/lib/python/rocpd/merge.py new file mode 100644 index 0000000000..0ccb2c1521 --- /dev/null +++ b/source/lib/python/rocpd/merge.py @@ -0,0 +1,363 @@ +import argparse +import sqlite3 +import uuid +import os +from typing import Any + +from .importer import RocpdImportData, execute_statement +from .schema import RocpdSchema +from .time_window import get_column_names + + +def create_empty_db(output_file, new_uuid, new_guid): + """ + Create an empty database with the schema. + + Returns: + Connection to the database + """ + output_dir = os.path.dirname(output_file) + if output_dir and not os.path.exists(output_dir): + os.makedirs(output_dir, exist_ok=True) + + if os.path.isfile(output_file): + os.remove(output_file) + + conn = sqlite3.connect(output_file) + + schema = RocpdSchema(uuid=new_uuid, guid=new_guid) + schema.write_schema(conn) + + conn.commit() + return RocpdImportData(output_file) + + +def get_all_db_uuids(import_data): + result = execute_statement(import_data, "PRAGMA database_list").fetchall() + all_db_uuids = [] + for db in result: + if db[1] in ["main", "temp"]: + continue + for itr in execute_statement( + import_data, + f"SELECT value FROM {db[1]}.rocpd_metadata WHERE tag='uuid'", + ).fetchall(): + all_db_uuids.append((db[1], itr[0])) + return all_db_uuids + + +def update_table_ids(import_data, alias, table, uid, max_id): + if max_id == 0: + return + + stmt = f"SELECT id FROM {alias}.{table}{uid} ORDER BY id DESC" + ids = execute_statement(import_data, stmt).fetchall() + + for (old_id,) in ids: + update_stmt = f""" + UPDATE {alias}.{table}{uid} + SET id = {old_id + max_id} + WHERE id = {old_id} + """ + execute_statement(import_data, update_stmt) + + import_data.commit() + + +def undo_update_table_ids(db_conn, alias, table, uid): + ids = execute_statement( + db_conn, f"SELECT id FROM {alias}.{table}{uid} ORDER BY id ASC" + ).fetchall() + for idx, (old_id,) in enumerate(ids): + update_stmt = f""" + UPDATE {alias}.{table}{uid} + SET id = {idx} + WHERE id = {old_id} + """ + execute_statement(db_conn, update_stmt) + db_conn.commit() + + +def insert_rocpd_info_node(connection, all_db_uuids, new_connection, new_uuid) -> None: + updates_needed = {} + unique_nodes = {} + + for alias, _uuid in all_db_uuids: + updates_needed[alias] = [] # TODO alias -> alias+ _uuid ? + rows = execute_statement( + connection, f"SELECT * FROM {alias}.rocpd_info_node{_uuid}" + ).fetchall() + for row in rows: + node_hash = row[2] # Hash value + node_id = row[0] + if node_hash not in unique_nodes: + unique_nodes[node_hash] = (node_id, row) # TODO hash -> hash+machine_id ? + + elif node_id != unique_nodes[node_hash][0]: + updates_needed[alias].append((node_id, unique_nodes[node_hash][0])) + + for alias, _uuid in all_db_uuids: + if updates_needed[alias]: + for old_id, new_id in updates_needed[alias]: + execute_statement( + connection, + f""" + UPDATE {alias}.rocpd_info_node{_uuid} + SET id = ? + WHERE id = ? + """, + (new_id, old_id), + ) + connection.commit() + + cur = new_connection.cursor() + for _, node in unique_nodes.values(): + # TODO + cur.execute( + f"INSERT OR IGNORE INTO rocpd_info_node{new_uuid} (id, guid, hash, machine_id, system_name, hostname, release, version, hardware_name, domain_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + node, + ) + new_connection.commit() + + +def insert_rocpd_string(connection, all_db_uuids, new_connection, new_uuid): + strings_mapping = {} + ids = [] + updates_needed = {} + + for alias, _uuid in all_db_uuids: + updates_needed[alias] = [] # TODO alias -> alias + _uuid ? + rows = connection.execute( + f"SELECT id, string FROM {alias}.rocpd_string{_uuid}" + ).fetchall() + for r in rows: + row_id = r[0] + row_string = r[1] + + if row_string not in strings_mapping.keys(): + if row_id not in ids: + strings_mapping[row_string] = row_id + ids.append(row_id) + else: + ids.sort() + new_id = ids[-1] + 1 + strings_mapping[row_string] = new_id + ids.append(new_id) + updates_needed[alias].append((new_id, row_id, row_string)) + + elif row_id != strings_mapping[row_string]: + updates_needed[alias].append( + (row_id, strings_mapping[row_string], row_string) + ) + + for alias, _uuid in all_db_uuids: + if updates_needed[alias]: + for old_id, new_id, string in updates_needed[alias]: + connection.execute( + f""" + UPDATE {alias}.rocpd_string{_uuid} + SET id = ? + WHERE id = ? + """, + (new_id, old_id), + ) + + connection.commit() + + cur = new_connection.cursor() + for string, id_value in strings_mapping.items(): + cur.execute( + f"INSERT INTO rocpd_string{new_uuid} (id, string) VALUES (?, ?)", + (id_value, string), + ) + new_connection.commit() + + +def insert_table(table, alias, uid, new_uuid, import_data, import_data_merge): + rows = execute_statement( + import_data, f"SELECT * FROM {alias}.{table}{uid}" + ).fetchall() + + column_names = get_column_names(import_data, table) + placeholders = ",".join(["?"] * len(column_names)) + insert_sql = f"INSERT INTO {table}{new_uuid} ({','.join(column_names)}) VALUES ({placeholders})" + + dest_cur = import_data_merge.cursor() + dest_cur.executemany(insert_sql, rows) + import_data_merge.commit() + + +def update_tables_new_guid(new_db_conn, new_guid): + cursor = new_db_conn.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") + table_names = [t[0] for t in cursor.fetchall() if not t[0].startswith("sqlite_")] + + # Update all guids in all tables with the new GUID + for t in table_names: + cursor.execute(f"PRAGMA table_info({t})") + cols = [col[1] for col in cursor.fetchall()] + if "guid" in cols: + new_db_conn.execute(f"UPDATE {t} SET guid = ?", (new_guid,)) + + new_db_conn.commit() + + +def post_process(new_import_data, new_uuid): + # Update agents absolute_index + agent_types = [ + row[0] + for row in execute_statement( + new_import_data, f"SELECT DISTINCT type FROM rocpd_info_agent{new_uuid}" + ).fetchall() + ] + ids = [ + row[0] + for row in execute_statement( + new_import_data, f"SELECT id FROM rocpd_info_agent{new_uuid} ORDER BY id" + ).fetchall() + ] + for id in ids: + execute_statement( + new_import_data, + f"UPDATE rocpd_info_agent{new_uuid} SET absolute_index = {id} WHERE id = {id}", + ) + + # Update agents Type index + for agent_type in agent_types: + ids = [ + row[0] + for row in execute_statement( + new_import_data, + f"SELECT id FROM rocpd_info_agent{new_uuid} WHERE type ='{agent_type}' ORDER BY id", + ).fetchall() + ] + + for new_type_index, agent_id in enumerate(ids): + execute_statement( + new_import_data, + f"UPDATE rocpd_info_agent{new_uuid} SET type_index = {new_type_index} WHERE id = {agent_id}", + ) + new_import_data.commit() + + +def merge(import_data: RocpdImportData, **kwargs: Any) -> None: + import time + + start_time = time.time() + + new_guid = str(uuid.uuid1()) + new_uuid = f"_{new_guid}".replace("-", "_") + + # Create an empty db in output_merge_path + output = kwargs.get("output_merge_path") + new_import_data = create_empty_db(output, new_uuid, new_guid) + + # List all dbs and their uuids + all_db_uuids = get_all_db_uuids(import_data) + + special_table_cases = ["rocpd_metadata", "rocpd_string", "rocpd_info_node"] + table_names = [ + t for t in import_data.table_info.keys() if t not in special_table_cases + ] + + # Update ids in orig connection + print("Updating ids in original database (this may take a while)...") + for table in table_names: + max_id = 0 + for alias, _uuid in all_db_uuids: + update_table_ids(import_data, alias, table, _uuid, max_id) + + new_max = execute_statement( + import_data, f"SELECT max(id) FROM {alias}.{table}{_uuid}" + ).fetchall()[0][0] + + if new_max: + max_id += new_max + 1 + + # Insert special cases rocpd_info_node + rocpd_string + print("Inserting special cases rocpd_info_node + rocpd_string...") + insert_rocpd_info_node(import_data, all_db_uuids, new_import_data, new_uuid) + insert_rocpd_string(import_data, all_db_uuids, new_import_data, new_uuid) + + # Insert rest of the data + print("Inserting data from all tables...") + for table in table_names: + for alias, _uuid in all_db_uuids: + insert_table(table, alias, _uuid, new_uuid, import_data, new_import_data) + + # Revert changes in original db + print("Reverting changes in original database (this may take a while)...") + for table in import_data.table_info.keys(): + for alias, _uuid in all_db_uuids: + undo_update_table_ids(import_data, alias, table, _uuid) + + # Update new guid + print("Updating GUID in new database...") + update_tables_new_guid(new_import_data, new_guid) + + # Post-process agents + post_process(new_import_data, new_uuid) + + elapsed_time = time.time() - start_time + print(f"Merge completed successfully! Output saved to: {output}") + print(f"Time: {elapsed_time:.2f} sec") + + +# +# Command-line interface functions +# +def add_args(parser): + """Add arguments for merger.""" + merge_options = parser.add_argument_group("Merge options") + merge_options.add_argument( + "--output-merge-path", + help="Sets the output path where the output merge files will be saved (default path: `./rocpd-output-data/db_merged.db`)", + default=os.environ.get("ROCPD_OUTPUT_PATH", "./rocpd-output-data/db_merged.db"), + type=str, + required=False, + ) + return ["output_merge_path"] + + +def process_args(args, valid_args): + ret = {} + for itr in valid_args: + if hasattr(args, itr): + val = getattr(args, itr) + if val is not None: + ret[itr] = val + return ret + + +def execute(input_rpd: str, **kwargs: Any) -> RocpdImportData: + + importData = RocpdImportData(input_rpd) + + merge(importData, **kwargs) + + return importData + + +def main(argv=None) -> int: + """Main entry point for command line execution.""" + + parser = argparse.ArgumentParser(description="Merge ROCpd databases") + parser.add_argument( + "-i", + "--input", + type=str, + required=True, + help="Path to the input ROCpd database files", + ) + + valid_args = add_args(parser) + + args = parser.parse_args(argv) + + merged_args = process_args(args, valid_args) + + execute(args.input, **merged_args) + + +if __name__ == "__main__": + main() diff --git a/source/lib/python/utilities.cmake b/source/lib/python/utilities.cmake index 9c8a987d76..4f56b96b15 100644 --- a/source/lib/python/utilities.cmake +++ b/source/lib/python/utilities.cmake @@ -151,6 +151,7 @@ function(rocprofiler_rocpd_python_bindings _VERSION) importer.py __init__.py __main__.py + merge.py output_config.py otf2.py pftrace.py From d7202602993fc8c43842715e83734a8ca76858d3 Mon Sep 17 00:00:00 2001 From: acanadas Date: Thu, 19 Jun 2025 14:06:44 +0000 Subject: [PATCH 2/5] adding ROCpd database merge concatenating all tables --- source/lib/python/rocpd/merge.py | 460 +++++++++++++------------------ 1 file changed, 185 insertions(+), 275 deletions(-) diff --git a/source/lib/python/rocpd/merge.py b/source/lib/python/rocpd/merge.py index 0ccb2c1521..c823644dd7 100644 --- a/source/lib/python/rocpd/merge.py +++ b/source/lib/python/rocpd/merge.py @@ -1,305 +1,214 @@ +#!/usr/bin/env python3 +############################################################################### +# MIT License +# +# Copyright (c) 2025 Advanced Micro Devices, Inc. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +############################################################################### + import argparse -import sqlite3 -import uuid import os -from typing import Any +import sqlite3 +import time +from collections import defaultdict +from pathlib import Path +from typing import List, Tuple, Any from .importer import RocpdImportData, execute_statement from .schema import RocpdSchema from .time_window import get_column_names +__all__ = ["RocpdMergeData", "merge"] -def create_empty_db(output_file, new_uuid, new_guid): - """ - Create an empty database with the schema. - - Returns: - Connection to the database - """ - output_dir = os.path.dirname(output_file) - if output_dir and not os.path.exists(output_dir): - os.makedirs(output_dir, exist_ok=True) - - if os.path.isfile(output_file): - os.remove(output_file) - - conn = sqlite3.connect(output_file) - - schema = RocpdSchema(uuid=new_uuid, guid=new_guid) - schema.write_schema(conn) - - conn.commit() - return RocpdImportData(output_file) - - -def get_all_db_uuids(import_data): - result = execute_statement(import_data, "PRAGMA database_list").fetchall() - all_db_uuids = [] - for db in result: - if db[1] in ["main", "temp"]: - continue - for itr in execute_statement( - import_data, - f"SELECT value FROM {db[1]}.rocpd_metadata WHERE tag='uuid'", - ).fetchall(): - all_db_uuids.append((db[1], itr[0])) - return all_db_uuids - - -def update_table_ids(import_data, alias, table, uid, max_id): - if max_id == 0: - return - - stmt = f"SELECT id FROM {alias}.{table}{uid} ORDER BY id DESC" - ids = execute_statement(import_data, stmt).fetchall() - - for (old_id,) in ids: - update_stmt = f""" - UPDATE {alias}.{table}{uid} - SET id = {old_id + max_id} - WHERE id = {old_id} - """ - execute_statement(import_data, update_stmt) - - import_data.commit() - - -def undo_update_table_ids(db_conn, alias, table, uid): - ids = execute_statement( - db_conn, f"SELECT id FROM {alias}.{table}{uid} ORDER BY id ASC" - ).fetchall() - for idx, (old_id,) in enumerate(ids): - update_stmt = f""" - UPDATE {alias}.{table}{uid} - SET id = {idx} - WHERE id = {old_id} - """ - execute_statement(db_conn, update_stmt) - db_conn.commit() - - -def insert_rocpd_info_node(connection, all_db_uuids, new_connection, new_uuid) -> None: - updates_needed = {} - unique_nodes = {} - - for alias, _uuid in all_db_uuids: - updates_needed[alias] = [] # TODO alias -> alias+ _uuid ? - rows = execute_statement( - connection, f"SELECT * FROM {alias}.rocpd_info_node{_uuid}" - ).fetchall() - for row in rows: - node_hash = row[2] # Hash value - node_id = row[0] - if node_hash not in unique_nodes: - unique_nodes[node_hash] = (node_id, row) # TODO hash -> hash+machine_id ? - - elif node_id != unique_nodes[node_hash][0]: - updates_needed[alias].append((node_id, unique_nodes[node_hash][0])) - - for alias, _uuid in all_db_uuids: - if updates_needed[alias]: - for old_id, new_id in updates_needed[alias]: - execute_statement( - connection, - f""" - UPDATE {alias}.rocpd_info_node{_uuid} - SET id = ? - WHERE id = ? - """, - (new_id, old_id), - ) - connection.commit() - - cur = new_connection.cursor() - for _, node in unique_nodes.values(): - # TODO - cur.execute( - f"INSERT OR IGNORE INTO rocpd_info_node{new_uuid} (id, guid, hash, machine_id, system_name, hostname, release, version, hardware_name, domain_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - node, - ) - new_connection.commit() - - -def insert_rocpd_string(connection, all_db_uuids, new_connection, new_uuid): - strings_mapping = {} - ids = [] - updates_needed = {} - - for alias, _uuid in all_db_uuids: - updates_needed[alias] = [] # TODO alias -> alias + _uuid ? - rows = connection.execute( - f"SELECT id, string FROM {alias}.rocpd_string{_uuid}" - ).fetchall() - for r in rows: - row_id = r[0] - row_string = r[1] - - if row_string not in strings_mapping.keys(): - if row_id not in ids: - strings_mapping[row_string] = row_id - ids.append(row_id) - else: - ids.sort() - new_id = ids[-1] + 1 - strings_mapping[row_string] = new_id - ids.append(new_id) - updates_needed[alias].append((new_id, row_id, row_string)) - - elif row_id != strings_mapping[row_string]: - updates_needed[alias].append( - (row_id, strings_mapping[row_string], row_string) - ) - for alias, _uuid in all_db_uuids: - if updates_needed[alias]: - for old_id, new_id, string in updates_needed[alias]: - connection.execute( - f""" - UPDATE {alias}.rocpd_string{_uuid} - SET id = ? - WHERE id = ? - """, - (new_id, old_id), - ) +def prepare_output_file(output: str) -> None: + """Prepare output file by creating directory and removing existing file""" - connection.commit() - - cur = new_connection.cursor() - for string, id_value in strings_mapping.items(): - cur.execute( - f"INSERT INTO rocpd_string{new_uuid} (id, string) VALUES (?, ?)", - (id_value, string), - ) - new_connection.commit() - - -def insert_table(table, alias, uid, new_uuid, import_data, import_data_merge): - rows = execute_statement( - import_data, f"SELECT * FROM {alias}.{table}{uid}" - ).fetchall() - - column_names = get_column_names(import_data, table) - placeholders = ",".join(["?"] * len(column_names)) - insert_sql = f"INSERT INTO {table}{new_uuid} ({','.join(column_names)}) VALUES ({placeholders})" - - dest_cur = import_data_merge.cursor() - dest_cur.executemany(insert_sql, rows) - import_data_merge.commit() - - -def update_tables_new_guid(new_db_conn, new_guid): - cursor = new_db_conn.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") - table_names = [t[0] for t in cursor.fetchall() if not t[0].startswith("sqlite_")] - - # Update all guids in all tables with the new GUID - for t in table_names: - cursor.execute(f"PRAGMA table_info({t})") - cols = [col[1] for col in cursor.fetchall()] - if "guid" in cols: - new_db_conn.execute(f"UPDATE {t} SET guid = ?", (new_guid,)) - - new_db_conn.commit() - - -def post_process(new_import_data, new_uuid): - # Update agents absolute_index - agent_types = [ - row[0] - for row in execute_statement( - new_import_data, f"SELECT DISTINCT type FROM rocpd_info_agent{new_uuid}" - ).fetchall() - ] - ids = [ - row[0] - for row in execute_statement( - new_import_data, f"SELECT id FROM rocpd_info_agent{new_uuid} ORDER BY id" - ).fetchall() - ] - for id in ids: - execute_statement( - new_import_data, - f"UPDATE rocpd_info_agent{new_uuid} SET absolute_index = {id} WHERE id = {id}", - ) - - # Update agents Type index - for agent_type in agent_types: - ids = [ - row[0] - for row in execute_statement( - new_import_data, - f"SELECT id FROM rocpd_info_agent{new_uuid} WHERE type ='{agent_type}' ORDER BY id", - ).fetchall() - ] - - for new_type_index, agent_id in enumerate(ids): - execute_statement( - new_import_data, - f"UPDATE rocpd_info_agent{new_uuid} SET type_index = {new_type_index} WHERE id = {agent_id}", - ) - new_import_data.commit() + output_path = Path(output) + # Create parent directory if needed + if output_path.parent != Path("."): + output_path.parent.mkdir(parents=True, exist_ok=True) -def merge(import_data: RocpdImportData, **kwargs: Any) -> None: - import time + # Remove existing file + if output_path.is_file(): + output_path.unlink() + + +def get_database_list(import_data: RocpdImportData) -> List[Tuple[int, str, str]]: + """Get list of all attached databases with their sequence numbers, names, and file""" + return execute_statement(import_data, "PRAGMA database_list").fetchall() - start_time = time.time() - new_guid = str(uuid.uuid1()) - new_uuid = f"_{new_guid}".replace("-", "_") +def get_table_names_per_alias(import_data: RocpdImportData, alias: str) -> List[str]: + """Get all table names from a specific database alias""" - # Create an empty db in output_merge_path - output = kwargs.get("output_merge_path") - new_import_data = create_empty_db(output, new_uuid, new_guid) + query = f"SELECT name FROM {alias}.sqlite_master WHERE type='table';" + rows = execute_statement(import_data, query).fetchall() + return [table[0] for table in rows if not table[0].startswith("sqlite_")] - # List all dbs and their uuids - all_db_uuids = get_all_db_uuids(import_data) - special_table_cases = ["rocpd_metadata", "rocpd_string", "rocpd_info_node"] - table_names = [ - t for t in import_data.table_info.keys() if t not in special_table_cases - ] +def get_table_data(import_data: RocpdImportData, table_name: str): + """Get table data using execute_statement""" + return execute_statement(import_data, f"SELECT * FROM {table_name}").fetchall() - # Update ids in orig connection - print("Updating ids in original database (this may take a while)...") - for table in table_names: - max_id = 0 - for alias, _uuid in all_db_uuids: - update_table_ids(import_data, alias, table, _uuid, max_id) - new_max = execute_statement( - import_data, f"SELECT max(id) FROM {alias}.{table}{_uuid}" - ).fetchall()[0][0] +def get_uuid_guid(import_data: RocpdImportData, metadata_table: str) -> Tuple[str, str]: + """Get UUID and GUID values from the metadata table""" + uuid = execute_statement( + import_data, + f"SELECT value FROM {metadata_table} WHERE tag='uuid' ORDER BY id ASC", + ).fetchone()[0] + guid = execute_statement( + import_data, + f"SELECT value FROM {metadata_table} WHERE tag='guid' ORDER BY id ASC", + ).fetchone()[0] + return uuid, guid - if new_max: - max_id += new_max + 1 - # Insert special cases rocpd_info_node + rocpd_string - print("Inserting special cases rocpd_info_node + rocpd_string...") - insert_rocpd_info_node(import_data, all_db_uuids, new_import_data, new_uuid) - insert_rocpd_string(import_data, all_db_uuids, new_import_data, new_uuid) +class RocpdMergeData: + """Utility class for merging ROCProfiler databases.""" + + def __init__(self, import_data: RocpdImportData, output_path: str): + if not isinstance(import_data, RocpdImportData): + raise ValueError( + f"Expected RocpdImportData, got {type(import_data).__name__}" + ) + + if not output_path: + raise ValueError("output_path cannot be empty") + + self.import_data = import_data + self.output_path = output_path + self._connection = None + + def __enter__(self): + """Support 'with RocpdMergeData(...) as merger:' pattern""" + prepare_output_file(self.output_path) + self._connection = sqlite3.connect(self.output_path) + return self + + def __exit__(self, exc_type, *_): + """Clean up resources""" + if self._connection: + try: + if exc_type is None: + self._connection.commit() + else: + self._connection.rollback() + finally: + self._connection.close() + self._connection = None + + return False + + def merge(self) -> None: + """Execute the merge operation""" + + all_tables = self._get_all_tables() + print(f"Merging {len(all_tables)} tables...") + + # Create tables per uuid / guid + uuid_guuids = [] + for table_name in all_tables: + if "rocpd_metadata" in table_name: + uuid, guid = get_uuid_guid(self.import_data, table_name) + self._connection.executescript(RocpdSchema(uuid=uuid, guid=guid).tables) + uuid_guuids.append((uuid, guid)) + + # Insert data in tables + for table_name in all_tables: + if "rocpd_metadata" in table_name: + continue + data = get_table_data(self.import_data, table_name) + if data: + column_names = get_column_names(self.import_data, table_name) + self._insert_data_into_merged(data, len(column_names), table_name) + + # Create rocpd_<> views + views_by_base_name = defaultdict(list) # view name -> list of table names + for _uuid, _ in uuid_guuids: + for tablename_with_uuid in all_tables: + if _uuid in tablename_with_uuid: + table_name = tablename_with_uuid.replace(_uuid, "") + views_by_base_name[table_name].append(tablename_with_uuid) + + self._connection.executescript(self._create_union_views(views_by_base_name)) + + # Create rest of the views + self._connection.executescript(RocpdSchema().views) + + def _get_all_tables(self) -> list: + """Get all tables from all attached databases and verify no duplicates exist""" + + dbs = get_database_list(self.import_data) + if len(dbs) <= 2: # main and temp + raise ValueError("No databases attached for merging") + + all_tables = [] + for db in dbs: + all_tables.extend(get_table_names_per_alias(self.import_data, db[1])) + + # Check for duplicates + unique_tables = set(all_tables) + assert len(all_tables) == len( + unique_tables + ), f"Duplicate tables found: {set([x for x in all_tables if all_tables.count(x) > 1])}" + + return all_tables + + def _insert_data_into_merged( + self, table_data: list, columns_count: int, table_name: str + ) -> None: + """Insert data into merged database""" + placeholders = ", ".join(["?" for _ in range(columns_count)]) + insert_statement = f"INSERT INTO {table_name} VALUES ({placeholders})" + self._connection.executemany(insert_statement, table_data) + + def _create_union_views(self, views_by_base_name) -> list: + union_views = [] + + for view_name, table_names in views_by_base_name.items(): + if len(table_names) == 1: + union_views.append( + f"""CREATE VIEW IF NOT EXISTS `{view_name}` AS SELECT * FROM `{table_names[0]}`;""" + ) + else: + select_statements = [f"SELECT * FROM `{table}`" for table in table_names] + union_query = "\nUNION ALL\n".join(select_statements) - # Insert rest of the data - print("Inserting data from all tables...") - for table in table_names: - for alias, _uuid in all_db_uuids: - insert_table(table, alias, _uuid, new_uuid, import_data, new_import_data) + union_views.append( + f"""CREATE VIEW IF NOT EXISTS `{view_name}` AS {union_query};""" + ) + return "\n\n".join(union_views) - # Revert changes in original db - print("Reverting changes in original database (this may take a while)...") - for table in import_data.table_info.keys(): - for alias, _uuid in all_db_uuids: - undo_update_table_ids(import_data, alias, table, _uuid) - # Update new guid - print("Updating GUID in new database...") - update_tables_new_guid(new_import_data, new_guid) +def merge(import_data: RocpdImportData, **kwargs: Any) -> None: + start_time = time.time() - # Post-process agents - post_process(new_import_data, new_uuid) + output_path = kwargs.get("output_merge_path") + with RocpdMergeData(import_data, output_path) as merger: + merger.merge() elapsed_time = time.time() - start_time - print(f"Merge completed successfully! Output saved to: {output}") + print(f"Merge completed successfully! Output saved to: {output_path}") print(f"Time: {elapsed_time:.2f} sec") @@ -316,6 +225,7 @@ def add_args(parser): type=str, required=False, ) + return ["output_merge_path"] From f6c701a5d039b528413c6b00243f6072dd53aff1 Mon Sep 17 00:00:00 2001 From: a-canadasruiz Date: Fri, 27 Jun 2025 13:23:05 +0200 Subject: [PATCH 3/5] update merge script - copy all tables from files --- source/lib/python/rocpd/__main__.py | 5 +- source/lib/python/rocpd/merge.py | 284 +++++++++++++--------------- 2 files changed, 132 insertions(+), 157 deletions(-) diff --git a/source/lib/python/rocpd/__main__.py b/source/lib/python/rocpd/__main__.py index 627d78a41d..46dfc8f9a6 100644 --- a/source/lib/python/rocpd/__main__.py +++ b/source/lib/python/rocpd/__main__.py @@ -291,10 +291,7 @@ def get_output_type(val): elif args.command == "merge": # merge subparser args merge_args = merge.process_args(args, valid_merge_args) - - # now start processing the data. Import the data and merge the views - importData = RocpdImportData(args.input) - merge.merge(importData, **merge_args) + merge.execute(args.input, **merge_args) # if the user requested query module, execute the query elif args.command == "query": diff --git a/source/lib/python/rocpd/merge.py b/source/lib/python/rocpd/merge.py index c823644dd7..3270cebb03 100644 --- a/source/lib/python/rocpd/merge.py +++ b/source/lib/python/rocpd/merge.py @@ -1,8 +1,7 @@ -#!/usr/bin/env python3 ############################################################################### # MIT License # -# Copyright (c) 2025 Advanced Micro Devices, Inc. +# Copyright (c) 2023 Advanced Micro Devices, Inc. # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal @@ -23,19 +22,22 @@ # THE SOFTWARE. ############################################################################### +# +# Utility classes to merge rpd files +# +# import argparse import os import sqlite3 import time + from collections import defaultdict +from typing import List, Any, Dict from pathlib import Path -from typing import List, Tuple, Any -from .importer import RocpdImportData, execute_statement -from .schema import RocpdSchema -from .time_window import get_column_names +# from .schema import RocpdSchema -__all__ = ["RocpdMergeData", "merge"] +__all__ = ["RocpdMerger", "execute"] def prepare_output_file(output: str) -> None: @@ -51,137 +53,37 @@ def prepare_output_file(output: str) -> None: if output_path.is_file(): output_path.unlink() +class RocpdMerger(): -def get_database_list(import_data: RocpdImportData) -> List[Tuple[int, str, str]]: - """Get list of all attached databases with their sequence numbers, names, and file""" - return execute_statement(import_data, "PRAGMA database_list").fetchall() - - -def get_table_names_per_alias(import_data: RocpdImportData, alias: str) -> List[str]: - """Get all table names from a specific database alias""" - - query = f"SELECT name FROM {alias}.sqlite_master WHERE type='table';" - rows = execute_statement(import_data, query).fetchall() - return [table[0] for table in rows if not table[0].startswith("sqlite_")] - - -def get_table_data(import_data: RocpdImportData, table_name: str): - """Get table data using execute_statement""" - return execute_statement(import_data, f"SELECT * FROM {table_name}").fetchall() - - -def get_uuid_guid(import_data: RocpdImportData, metadata_table: str) -> Tuple[str, str]: - """Get UUID and GUID values from the metadata table""" - uuid = execute_statement( - import_data, - f"SELECT value FROM {metadata_table} WHERE tag='uuid' ORDER BY id ASC", - ).fetchone()[0] - guid = execute_statement( - import_data, - f"SELECT value FROM {metadata_table} WHERE tag='guid' ORDER BY id ASC", - ).fetchone()[0] - return uuid, guid - - -class RocpdMergeData: - """Utility class for merging ROCProfiler databases.""" - - def __init__(self, import_data: RocpdImportData, output_path: str): - if not isinstance(import_data, RocpdImportData): + def __init__(self, input, output): + + if isinstance(input, sqlite3.Connection): raise ValueError( - f"Expected RocpdImportData, got {type(import_data).__name__}" + "RocpdMerger does not accept existing sqlite3 connections" + ) + elif isinstance(input, str): + raise ValueError( + "RocpdMerger only accepts a list of filenames to merge, not a single filename" + ) + elif isinstance(input, list) and len(input) > 0 and isinstance(input[0], str): + self._filenames = input[:] + self._output = output + prepare_output_file(self._output) + self._connection = sqlite3.connect(self._output) + + else: + raise ValueError( + f"input is unsupported type. Expected list of strings. type={type(input).__name__}" ) - - if not output_path: - raise ValueError("output_path cannot be empty") - - self.import_data = import_data - self.output_path = output_path - self._connection = None def __enter__(self): - """Support 'with RocpdMergeData(...) as merger:' pattern""" - prepare_output_file(self.output_path) - self._connection = sqlite3.connect(self.output_path) + # support "with RocpdMerge(...) as db:": return self - def __exit__(self, exc_type, *_): - """Clean up resources""" - if self._connection: - try: - if exc_type is None: - self._connection.commit() - else: - self._connection.rollback() - finally: - self._connection.close() - self._connection = None - - return False - - def merge(self) -> None: - """Execute the merge operation""" - - all_tables = self._get_all_tables() - print(f"Merging {len(all_tables)} tables...") - - # Create tables per uuid / guid - uuid_guuids = [] - for table_name in all_tables: - if "rocpd_metadata" in table_name: - uuid, guid = get_uuid_guid(self.import_data, table_name) - self._connection.executescript(RocpdSchema(uuid=uuid, guid=guid).tables) - uuid_guuids.append((uuid, guid)) - - # Insert data in tables - for table_name in all_tables: - if "rocpd_metadata" in table_name: - continue - data = get_table_data(self.import_data, table_name) - if data: - column_names = get_column_names(self.import_data, table_name) - self._insert_data_into_merged(data, len(column_names), table_name) - - # Create rocpd_<> views - views_by_base_name = defaultdict(list) # view name -> list of table names - for _uuid, _ in uuid_guuids: - for tablename_with_uuid in all_tables: - if _uuid in tablename_with_uuid: - table_name = tablename_with_uuid.replace(_uuid, "") - views_by_base_name[table_name].append(tablename_with_uuid) - - self._connection.executescript(self._create_union_views(views_by_base_name)) - - # Create rest of the views - self._connection.executescript(RocpdSchema().views) - - def _get_all_tables(self) -> list: - """Get all tables from all attached databases and verify no duplicates exist""" - - dbs = get_database_list(self.import_data) - if len(dbs) <= 2: # main and temp - raise ValueError("No databases attached for merging") - - all_tables = [] - for db in dbs: - all_tables.extend(get_table_names_per_alias(self.import_data, db[1])) - - # Check for duplicates - unique_tables = set(all_tables) - assert len(all_tables) == len( - unique_tables - ), f"Duplicate tables found: {set([x for x in all_tables if all_tables.count(x) > 1])}" - - return all_tables - - def _insert_data_into_merged( - self, table_data: list, columns_count: int, table_name: str - ) -> None: - """Insert data into merged database""" - placeholders = ", ".join(["?" for _ in range(columns_count)]) - insert_statement = f"INSERT INTO {table_name} VALUES ({placeholders})" - self._connection.executemany(insert_statement, table_data) - + def __exit__(self, exc_type, exc_value, traceback): + self._connection.close() + print('Closing connection to output database') + def _create_union_views(self, views_by_base_name) -> list: union_views = [] @@ -199,34 +101,101 @@ def _create_union_views(self, views_by_base_name) -> list: ) return "\n\n".join(union_views) - -def merge(import_data: RocpdImportData, **kwargs: Any) -> None: - start_time = time.time() - - output_path = kwargs.get("output_merge_path") - with RocpdMergeData(import_data, output_path) as merger: - merger.merge() - - elapsed_time = time.time() - start_time - print(f"Merge completed successfully! Output saved to: {output_path}") - print(f"Time: {elapsed_time:.2f} sec") - + def merge(self): + """ + Merge multiple SQLite databases into a single destination database. + """ + cur_dest = self._connection.cursor() + + views_by_base_name = defaultdict(list) + + versions = [] # Check that all databases have the same schema version + + for orig in self._filenames: + + print(f'Adding {orig}') + + con_orig = sqlite3.connect(orig) + cur_orig = con_orig.cursor() + + cur_orig.execute("SELECT name FROM sqlite_master WHERE type='table';") + tables = cur_orig.fetchall() + print(f'Tables found: {len(tables) -1}') + + uudid_statement = "SELECT value FROM rocpd_metadata WHERE tag='uuid'" + _uuid = [ itr[0] for itr in cur_orig.execute(uudid_statement).fetchall()][0] + + version_statement = "SELECT value FROM rocpd_metadata WHERE tag='schema_version'" + versions.extend([ itr[0] for itr in cur_orig.execute(version_statement).fetchall()]) + + for table in tables: + table_name = table[0] + if "sqlite_sequence" in table_name: + continue + + view_name = table_name.replace(_uuid, "") + views_by_base_name[view_name].append(table_name) + + cur_orig.execute(f"SELECT sql FROM sqlite_master WHERE type='table' AND name='{table_name}'") + create_table_stmt = cur_orig.fetchone()[0] + + cur_dest.execute(create_table_stmt) + + cur_orig.execute(f"SELECT * FROM {table_name}") + rows = cur_orig.fetchall() + for row in rows: + placeholders = ', '.join('?' * len(row)) + cur_dest.execute(f"INSERT INTO {table_name} VALUES ({placeholders})", row) + + con_orig.close() + + assert len(list(set(versions))) == 1 , f'Multiple versions found : {list(set(versions))}' + + # Create rocpd_<> views + self._connection.executescript(self._create_union_views(views_by_base_name)) + + # Create data views + con_orig = sqlite3.connect(self._filenames[0]) + orig_cursor = con_orig.cursor() + orig_cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='view' AND name NOT LIKE 'rocpd_%';") + views = orig_cursor.fetchall() + + for view in views: + _ , sql_view = view + self._connection.executescript(sql_view) + + con_orig.close() + # self._connection.executescript(RocpdSchema().views) + + self._connection.commit() # # Command-line interface functions # def add_args(parser): """Add arguments for merger.""" - merge_options = parser.add_argument_group("Merge options") - merge_options.add_argument( - "--output-merge-path", - help="Sets the output path where the output merge files will be saved (default path: `./rocpd-output-data/db_merged.db`)", - default=os.environ.get("ROCPD_OUTPUT_PATH", "./rocpd-output-data/db_merged.db"), + + o_options = parser.add_argument_group("Output options") + + o_options.add_argument( + "-o", + "--output-file", + help="Sets the base output file name", + default=os.environ.get("ROCPD_OUTPUT_NAME", "merged"), + type=str, + required=False, + ) + o_options.add_argument( + "-d", + "--output-path", + help="Sets the output path where the output files will be saved (default path: `./rocpd-output-data`)", + default=os.environ.get("ROCPD_OUTPUT_PATH", "./rocpd-output-data"), type=str, required=False, ) - return ["output_merge_path"] + return ["output_file", "output_path"] + def process_args(args, valid_args): @@ -239,13 +208,22 @@ def process_args(args, valid_args): return ret -def execute(input_rpd: str, **kwargs: Any) -> RocpdImportData: +def execute(inputs: List[str], **kwargs: Dict[str, Any]) -> None: + + start_time = time.time() - importData = RocpdImportData(input_rpd) + output_path = kwargs.get("output_path") + output_filename = kwargs.get("output_file") + ".db" + output = Path(output_path, output_filename) + + with RocpdMerger(inputs, output) as merger: + merger.merge() - merge(importData, **kwargs) + elapsed_time = time.time() - start_time + + print(f"Merge completed successfully! Output saved to: {output}") + print(f"Time: {elapsed_time:.2f} sec") - return importData def main(argv=None) -> int: From 9f2f6206207a51ec3058a0319fb57d7e1829c778 Mon Sep 17 00:00:00 2001 From: a-canadasruiz Date: Fri, 27 Jun 2025 13:25:16 +0200 Subject: [PATCH 4/5] fix merge format --- source/lib/python/rocpd/merge.py | 93 +++++++++++++++++--------------- 1 file changed, 51 insertions(+), 42 deletions(-) diff --git a/source/lib/python/rocpd/merge.py b/source/lib/python/rocpd/merge.py index 3270cebb03..b710789659 100644 --- a/source/lib/python/rocpd/merge.py +++ b/source/lib/python/rocpd/merge.py @@ -53,16 +53,15 @@ def prepare_output_file(output: str) -> None: if output_path.is_file(): output_path.unlink() -class RocpdMerger(): + +class RocpdMerger: def __init__(self, input, output): - + if isinstance(input, sqlite3.Connection): - raise ValueError( - "RocpdMerger does not accept existing sqlite3 connections" - ) + raise ValueError("RocpdMerger does not accept existing sqlite3 connections") elif isinstance(input, str): - raise ValueError( + raise ValueError( "RocpdMerger only accepts a list of filenames to merge, not a single filename" ) elif isinstance(input, list) and len(input) > 0 and isinstance(input[0], str): @@ -70,7 +69,7 @@ def __init__(self, input, output): self._output = output prepare_output_file(self._output) self._connection = sqlite3.connect(self._output) - + else: raise ValueError( f"input is unsupported type. Expected list of strings. type={type(input).__name__}" @@ -82,8 +81,7 @@ def __enter__(self): def __exit__(self, exc_type, exc_value, traceback): self._connection.close() - print('Closing connection to output database') - + def _create_union_views(self, views_by_base_name) -> list: union_views = [] @@ -107,27 +105,31 @@ def merge(self): """ cur_dest = self._connection.cursor() - views_by_base_name = defaultdict(list) - - versions = [] # Check that all databases have the same schema version - + views_by_base_name = defaultdict(list) + + versions = [] # Check that all databases have the same schema version + for orig in self._filenames: - - print(f'Adding {orig}') - + + print(f"Adding {orig}") + con_orig = sqlite3.connect(orig) cur_orig = con_orig.cursor() cur_orig.execute("SELECT name FROM sqlite_master WHERE type='table';") tables = cur_orig.fetchall() - print(f'Tables found: {len(tables) -1}') - + print(f"Tables found: {len(tables) -1}") + uudid_statement = "SELECT value FROM rocpd_metadata WHERE tag='uuid'" - _uuid = [ itr[0] for itr in cur_orig.execute(uudid_statement).fetchall()][0] - - version_statement = "SELECT value FROM rocpd_metadata WHERE tag='schema_version'" - versions.extend([ itr[0] for itr in cur_orig.execute(version_statement).fetchall()]) - + _uuid = [itr[0] for itr in cur_orig.execute(uudid_statement).fetchall()][0] + + version_statement = ( + "SELECT value FROM rocpd_metadata WHERE tag='schema_version'" + ) + versions.extend( + [itr[0] for itr in cur_orig.execute(version_statement).fetchall()] + ) + for table in tables: table_name = table[0] if "sqlite_sequence" in table_name: @@ -136,7 +138,9 @@ def merge(self): view_name = table_name.replace(_uuid, "") views_by_base_name[view_name].append(table_name) - cur_orig.execute(f"SELECT sql FROM sqlite_master WHERE type='table' AND name='{table_name}'") + cur_orig.execute( + f"SELECT sql FROM sqlite_master WHERE type='table' AND name='{table_name}'" + ) create_table_stmt = cur_orig.fetchone()[0] cur_dest.execute(create_table_stmt) @@ -144,37 +148,44 @@ def merge(self): cur_orig.execute(f"SELECT * FROM {table_name}") rows = cur_orig.fetchall() for row in rows: - placeholders = ', '.join('?' * len(row)) - cur_dest.execute(f"INSERT INTO {table_name} VALUES ({placeholders})", row) - + placeholders = ", ".join("?" * len(row)) + cur_dest.execute( + f"INSERT INTO {table_name} VALUES ({placeholders})", row + ) + con_orig.close() - - assert len(list(set(versions))) == 1 , f'Multiple versions found : {list(set(versions))}' - + + assert ( + len(list(set(versions))) == 1 + ), f"Multiple versions found : {list(set(versions))}" + # Create rocpd_<> views self._connection.executescript(self._create_union_views(views_by_base_name)) - + # Create data views con_orig = sqlite3.connect(self._filenames[0]) orig_cursor = con_orig.cursor() - orig_cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='view' AND name NOT LIKE 'rocpd_%';") + orig_cursor.execute( + "SELECT name, sql FROM sqlite_master WHERE type='view' AND name NOT LIKE 'rocpd_%';" + ) views = orig_cursor.fetchall() - - for view in views: - _ , sql_view = view + + for view in views: + _, sql_view = view self._connection.executescript(sql_view) - + con_orig.close() # self._connection.executescript(RocpdSchema().views) - + self._connection.commit() + # # Command-line interface functions # def add_args(parser): """Add arguments for merger.""" - + o_options = parser.add_argument_group("Output options") o_options.add_argument( @@ -197,7 +208,6 @@ def add_args(parser): return ["output_file", "output_path"] - def process_args(args, valid_args): ret = {} for itr in valid_args: @@ -215,17 +225,16 @@ def execute(inputs: List[str], **kwargs: Dict[str, Any]) -> None: output_path = kwargs.get("output_path") output_filename = kwargs.get("output_file") + ".db" output = Path(output_path, output_filename) - + with RocpdMerger(inputs, output) as merger: merger.merge() elapsed_time = time.time() - start_time - + print(f"Merge completed successfully! Output saved to: {output}") print(f"Time: {elapsed_time:.2f} sec") - def main(argv=None) -> int: """Main entry point for command line execution.""" From 76d8dd034cd9339c4794eb38e289bb7da292d092 Mon Sep 17 00:00:00 2001 From: Young Hui Date: Mon, 28 Jul 2025 23:12:58 -0400 Subject: [PATCH 5/5] Add package submodule, initial POC. Need to refine --- source/lib/python/rocpd/__main__.py | 54 ++++++- source/lib/python/rocpd/package.py | 222 ++++++++++++++++++++++++++++ source/lib/python/utilities.cmake | 1 + 3 files changed, 273 insertions(+), 4 deletions(-) create mode 100644 source/lib/python/rocpd/package.py diff --git a/source/lib/python/rocpd/__main__.py b/source/lib/python/rocpd/__main__.py index 46dfc8f9a6..689d0fb2ee 100644 --- a/source/lib/python/rocpd/__main__.py +++ b/source/lib/python/rocpd/__main__.py @@ -42,6 +42,7 @@ def main(argv=None, config=None): from . import merge from . import otf2 from . import output_config + from . import package from . import pftrace from . import query from . import summary @@ -74,6 +75,13 @@ def main(argv=None, config=None): TODO: Add examples for merge command """ + package_examples = """ + +Example usage: + + TODO: Add examples for package command +""" + query_examples = """ Example usage: @@ -134,6 +142,14 @@ def main(argv=None, config=None): epilog=merge_examples, ) + packager = subparsers.add_parser( + "package", + description="Package database files into .rpdb output", + allow_abbrev=False, + formatter_class=argparse.RawTextHelpFormatter, + epilog=package_examples, + ) + query_reporter = subparsers.add_parser( "query", description="Generate output on a query", @@ -184,6 +200,16 @@ def get_output_type(val): help="Input path and filename to one or more database(s)", ) + packager_required_params = packager.add_argument_group("Required options") + packager_required_params.add_argument( + "-i", + "--input", + required=True, + type=output_config.check_file_exists, + nargs="+", + help="Input path and filename to one or more database(s)", + ) + query_required_params = query_reporter.add_argument_group("Required options") query_required_params.add_argument( "-i", @@ -215,6 +241,9 @@ def get_output_type(val): # merge: subparser args valid_merge_args = merge.add_args(merger) + # package: subparser args + valid_package_args = package.add_args(packager) + # query: subparser args valid_out_config_args = output_config.add_args(query_reporter) valid_query_args = query.add_args(query_reporter) @@ -240,6 +269,17 @@ def get_output_type(val): parser.print_help() return + # convert to real number of DB input files + input_files = package.flatten_rocpd_yaml_input_file(args.input) + db_count = len(input_files) + + # TODO: add logic to determine how many DBs to merge into + ## SQLITE_MAX_ATTACHED == 10, can query once you have connection + ## if db_count > 10 + ## call merge to combine to fewer DBs + ## optionally, can also package up into own .rpdb package + ## Only after DBs have been merged and < SQLITE_MAX_ATTACHED, then we can call importer to attach DBs for analysis/convert + # if the user requested converter, process the conversion if args.command == "convert": # process the args @@ -253,7 +293,7 @@ def get_output_type(val): window_args = time_window.process_args(args, valid_time_window_args) # now start processing the data. Import the data and merge the views - importData = RocpdImportData(args.input) + importData = RocpdImportData(input_files) # adjust the time window view of the data if window_args is not None: @@ -291,7 +331,13 @@ def get_output_type(val): elif args.command == "merge": # merge subparser args merge_args = merge.process_args(args, valid_merge_args) - merge.execute(args.input, **merge_args) + merge.execute(input_files, **merge_args) + + # if the user requested package module, package up the database + elif args.command == "package": + # merge subparser args + package_args = package.process_args(args, valid_package_args) + package.execute(input_files, **package_args) # if the user requested query module, execute the query elif args.command == "query": @@ -303,7 +349,7 @@ def get_output_type(val): all_args = {**query_args, **out_cfg_args} query.execute( - args.input, + input_files, args, window_args=window_args, **all_args, @@ -317,7 +363,7 @@ def get_output_type(val): window_args = time_window.process_args(args, valid_time_window_args) # now start processing the data. Import the data and merge the views - importData = RocpdImportData(args.input) + importData = RocpdImportData(input_files) # adjust the time window view of the data if window_args is not None: diff --git a/source/lib/python/rocpd/package.py b/source/lib/python/rocpd/package.py new file mode 100644 index 0000000000..0c633e0932 --- /dev/null +++ b/source/lib/python/rocpd/package.py @@ -0,0 +1,222 @@ +import os + +# import json +import shutil +import datetime +import yaml +import argparse +from . import output_config + +rocpd_package_version = "1.0" + +rocpd_metadata_param_version = "rocpd_package_version" +rocpd_metadata_param_current_directory = "rocpd_current_directory" +rocpd_metadata_param_database_files = "rocpd_relative_path_to_database_files" + + +def flatten_rocpd_yaml_input_file(input) -> list: + """ + Processes a YAML file containing rocPD metadata and returns a list of database files. + + Args: + input (str): Path to the YAML file. + + Returns: + list: List of database file paths. + """ + # Flatten input list if any YAML file is provided + input_files = [] + for item in input: + if item.endswith((".yaml", ".yml")): + with open(item, "r") as f: + meta = yaml.safe_load(f) + cwd = meta.get(rocpd_metadata_param_current_directory, os.getcwd()) + dbs = meta.get(rocpd_metadata_param_database_files, []) + new_relative_dbs = [ + os.path.join(cwd, db) if not os.path.isabs(db) else db for db in dbs + ] + input_files.extend(new_relative_dbs) + else: + input_files.append(item) + return input_files + + +def create_metadata_file( + db_files, output_path=".", metadata_filename="index.yaml", consolidate=False +): + """ + Creates a metadata file listing the relative paths to the provided SQL database files. + + Args: + db_files (list of str): List of absolute or relative paths to SQL database files. + output_path (str): Directory to write the metadata file. + metadata_filename (str): Name of the metadata file to create. + + Returns: + str: Path to the created metadata file. + """ + # Ensure output directory exists + os.makedirs(output_path, exist_ok=True) + + # Compute relative paths + rel_paths = [os.path.relpath(db_file, output_path) for db_file in db_files] + + # If consolidating, set current directory to . + if consolidate: + current_directory = "." + else: + current_directory = os.path.normpath(os.path.join(os.getcwd(), output_path)) + + metadata = { + rocpd_metadata_param_version: rocpd_package_version, + rocpd_metadata_param_current_directory: current_directory, + rocpd_metadata_param_database_files: rel_paths, + } + + metadata_path = os.path.join(output_path, metadata_filename) + with open(metadata_path, "w") as f: + # json.dump(metadata, f, indent=4) # Uncomment for JSON format + yaml.safe_dump(metadata, f, default_flow_style=False) + + return metadata_path + + +def add_args(parser): + """Add arguments for package.""" + + package_options = parser.add_argument_group("Package options") + + package_options.add_argument( + "-c", + "--consolidate", + action="store_true", + help="Consolidate (copy) database files into a new folder and generate metadata file pointing to that folder", + ) + + package_options.add_argument( + "-d", + "--output-path", + help="Sets the name of output folder (default : current directory)", + # default=os.environ.get("ROCPD_OUTPUT_PATH", "./rocpd-output-data"), + type=str, + required=False, + ) + + return [ + "consolidate", + "output_path", + ] + + +def process_args(args, valid_args): + + ret = {} + for itr in valid_args: + if hasattr(args, itr): + val = getattr(args, itr) + if val is not None: + ret[itr] = val + return ret + + +def execute(input_files, **kwargs): + + output_path = kwargs.get("output_path", ".") + consolidate = kwargs.get("consolidate", "False") + + # Create a new folder with current date and time for unique folder to consolidate files to + if consolidate: + date_str = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + output_path = f"rocpd-{date_str}.rpdb" + + if consolidate: + # Create a new folder with current date and time + os.makedirs(output_path, exist_ok=True) + copied_files = [] + for db_file in input_files: + dest_file = os.path.join(output_path, os.path.basename(db_file)) + # Only copy if source and destination are not the same file + if os.path.abspath(db_file) != os.path.abspath(dest_file): + shutil.copy2(db_file, dest_file) + copied_files.append(dest_file) + metadata_path = create_metadata_file(copied_files, output_path, consolidate=True) + else: + # If not consolidating, just create metadata file with relative paths to current directory + metadata_path = create_metadata_file(input_files, output_path) + + print(f"rocPD package created at: {metadata_path}") + + +def main(argv=None): + """ + Main function to demonstrate the creation of a metadata file. + Supports copying database files to a new folder if --copy-db is specified. + """ + + parser = argparse.ArgumentParser( + description="Convert rocPD to Perfetto file", allow_abbrev=False + ) + + required_params = parser.add_argument_group("Required options") + + required_params.add_argument( + "-i", + "--input", + required=True, + type=output_config.check_file_exists, + nargs="+", + help="Input path and filename to one or more database(s), separated by spaces", + ) + + parser.add_argument( + "-d", + "--output-path", + help="Sets the name of output folder (default : current directory)", + # default=os.environ.get("ROCPD_OUTPUT_PATH", "./rocpd-output-data"), + type=str, + required=False, + ) + + parser.add_argument( + "-c", + "--consolidate", + action="store_true", + help="Consolidate (copy) database files into a new folder and generate metadata file pointing to that folder", + ) + + args = parser.parse_args(argv) + + input_files = flatten_rocpd_yaml_input_file(args.input) + + # TODO: fix this complicated logic. Let's make it simpler. + if args.output_path: + output_path = args.output_path + else: + # Create a new folder with current date and time for unique folder to consolidate files to + if args.consolidate: + date_str = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + output_path = f"rocpd-{date_str}.rpdb" + else: + output_path = "." + + if args.consolidate: + # Create a new folder with current date and time + os.makedirs(output_path, exist_ok=True) + copied_files = [] + for db_file in input_files: + dest_file = os.path.join(output_path, os.path.basename(db_file)) + # Only copy if source and destination are not the same file + if os.path.abspath(db_file) != os.path.abspath(dest_file): + shutil.copy2(db_file, dest_file) + copied_files.append(dest_file) + metadata_path = create_metadata_file(copied_files, output_path, consolidate=True) + else: + # If not consolidating, just create metadata file with relative paths to current directory + metadata_path = create_metadata_file(input_files, output_path) + + print(f"rocPD package created at: {metadata_path}") + + +# This is the entry point for the script. +if __name__ == "__main__": + main() diff --git a/source/lib/python/utilities.cmake b/source/lib/python/utilities.cmake index 4f56b96b15..b13722d3a0 100644 --- a/source/lib/python/utilities.cmake +++ b/source/lib/python/utilities.cmake @@ -154,6 +154,7 @@ function(rocprofiler_rocpd_python_bindings _VERSION) merge.py output_config.py otf2.py + package.py pftrace.py query.py schema.py