diff --git a/cmflib/cmf.py b/cmflib/cmf.py index b45fcc0fb..44f9a4da9 100644 --- a/cmflib/cmf.py +++ b/cmflib/cmf.py @@ -47,6 +47,7 @@ from cmflib import graph_wrapper from cmflib.store.sqllite_store import SqlliteStore from cmflib.store.postgres import PostgresStore +from cmflib.store.custom_sqlite_store import CustomSqliteStore from cmflib.metadata_helper import ( get_or_create_parent_context, get_or_create_run_context, @@ -169,6 +170,9 @@ def __init__( cur_folder = os.path.basename(os.getcwd()) pipeline_name = cur_folder self.pipeline_name = pipeline_name + self.custom_store = None + if is_server is False: + self.custom_store = CustomSqliteStore(filepath) self.store = temp_store.connect() self.filepath = filepath self.child_context = None @@ -487,7 +491,7 @@ def create_execution( #print(f"{python_env_file_path} doesn't exists!!") with open(python_env_file_path, 'w') as file: file.write(env_output) - + if self.graph: self.driver.create_execution_node( self.execution_name, @@ -759,7 +763,7 @@ def log_dataset( # If the dataset already exist , then we just link the existing dataset to the execution # We do not update the dataset properties . # We need to append the new properties to the existing dataset properties - custom_props = {} if custom_properties is None else custom_properties + custom_props = {} if custom_properties is None else dict(custom_properties) git_repo = git_get_repo() name = re.split("/", url)[-1] @@ -779,7 +783,7 @@ def log_dataset( dvc_url = dvc_get_url(url) dvc_url_with_pipeline = f"{self.parent_context.name}:{dvc_url}" url = url + ":" + c_hash - if c_hash and c_hash.strip: + if c_hash and c_hash.strip(): existing_artifact.extend(self.store.get_artifacts_by_uri(c_hash)) uri = c_hash @@ -829,8 +833,8 @@ def log_dataset( milliseconds_since_epoch=int(time.time() * 1000), ) - custom_props["git_repo"] = git_repo - custom_props["Commit"] = dataset_commit + system_props = {"git_repo": git_repo, "Commit": dataset_commit} + graph_custom_props = {**custom_props, **system_props} self.execution_label_props["git_repo"] = git_repo self.execution_label_props["Commit"] = dataset_commit @@ -842,7 +846,7 @@ def log_dataset( event, self.execution.id, self.parent_context, - custom_props, + graph_custom_props, ) if event.lower() == "input": self.input_artifacts.append( @@ -878,6 +882,29 @@ def log_dataset( if label: self.log_label(label, artifact_path, label_properties) + execution_uuid = self.execution.properties["Execution_uuid"].string_value.split(",")[-1].strip() + if self.custom_store and execution_uuid and artifact: + try: + # Step 1: Get artifact uri and build artifact-level metadata payload. + artifact_uri = str(getattr(artifact, "uri", "")).strip() + dataset_log_payload = { + "name": name, + "properties": { + "git_repo": git_repo, + "url": dvc_url_with_pipeline, + }, + "custom_properties": custom_props, + } + # Step 2: Upsert one ExecutionLogs row keyed by (execution_uuid, artifact_uri). + if artifact_uri: + self.custom_store.insert_execution_log( + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + metadata=dataset_log_payload, + ) + except Exception as exc: + logger.warning("Failed to write dataset metadata into execution_log: %s", exc) + os.chdir(logging_dir) return artifact @@ -1118,6 +1145,32 @@ def log_model( self.driver.create_artifact_relationships( self.input_artifacts, child_artifact, self.execution_label_props ) + execution_uuid = self.execution.properties["Execution_uuid"].string_value.split(",")[-1].strip() + if self.custom_store and execution_uuid and artifact: + try: + # Step 1: Get artifact uri and build artifact-level metadata payload. + artifact_uri = str(getattr(artifact, "uri", "")).strip() + model_log_payload = { + "name": model_name, + "properties": { + "model_framework": model_framework, + "model_type": model_type, + "model_name": model_name, + "Commit": model_commit, + "url": url_with_pipeline, + }, + "custom_properties": custom_props, + } + # Step 2: Upsert one ExecutionLogs row keyed by (execution_uuid, artifact_uri). + if artifact_uri: + self.custom_store.insert_execution_log( + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + metadata=model_log_payload, + ) + except Exception as exc: + logger.warning("Failed to write model metadata into execution_log: %s", exc) + os.chdir(logging_dir) return artifact @@ -1158,7 +1211,8 @@ def log_execution_metrics( self.create_execution(execution_type=assigned_name) assert self.execution is not None, f"Failed to create execution for {self.pipeline_name}!!" - custom_props = {} if custom_properties is None else custom_properties + custom_props = {} if custom_properties is None else dict(custom_properties) + coarse_metrics_name = metrics_name uri = str(uuid.uuid1()) metrics_name = metrics_name + ":" + uri + ":" + str(self.execution.id) metrics = create_new_artifact_event_and_attribution( @@ -1197,6 +1251,29 @@ def log_execution_metrics( self.driver.create_artifact_relationships( self.input_artifacts, child_artifact, self.execution_label_props ) + + execution_uuid = self.execution.properties["Execution_uuid"].string_value.split(",")[-1].strip() + if self.custom_store and execution_uuid and metrics: + try: + # Step 1: Get artifact uri and build artifact-level metadata payload. + artifact_uri = str(getattr(metrics, "uri", "")).strip() + metrics_log_payload = { + "name": coarse_metrics_name, + "properties": { + "metrics_name": metrics_name, + }, + "custom_properties": custom_props, + } + # Step 2: Upsert one ExecutionLogs row keyed by (execution_uuid, artifact_uri). + if artifact_uri: + self.custom_store.insert_execution_log( + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + metadata=metrics_log_payload, + ) + except Exception as exc: + logger.warning("Failed to write execution metrics metadata into execution_log: %s", exc) + os.chdir(logging_dir) return metrics @@ -1336,6 +1413,30 @@ def commit_metrics(self, metrics_name: str): self.input_artifacts, child_artifact, self.execution_label_props ) + execution_uuid = self.execution.properties["Execution_uuid"].string_value.split(",")[-1].strip() + if self.custom_store and execution_uuid and metrics: + try: + # Step 1: Get artifact uri and build artifact-level metadata payload. + artifact_uri = str(getattr(metrics, "uri", "")).strip() + step_metrics_log_payload = { + "name": metrics_name, + "properties": { + "Commit": metrics_commit, + "url": dvc_url_with_pipeline, + "metrics_path": metrics_path, + }, + "custom_properties": {}, + } + # Step 2: Upsert one ExecutionLogs row keyed by (execution_uuid, artifact_uri). + if artifact_uri: + self.custom_store.insert_execution_log( + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + metadata=step_metrics_log_payload, + ) + except Exception as exc: + logger.warning("Failed to write step metrics metadata into execution_log: %s", exc) + os.chdir(logging_dir) return metrics @@ -1495,7 +1596,7 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t # If the dataset already exist , then we just link the existing dataset to the execution # We do not update the dataset properties . # We need to append the new properties to the existing dataset properties - custom_props = {} if custom_properties is None else custom_properties + custom_props = {} if custom_properties is None else dict(custom_properties) git_repo = git_get_repo() name = re.split("/", url)[-1] @@ -1506,7 +1607,7 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t # Calculate label_hash label_hash = calculate_md5(url) - # Get dataset_uri from DVC + # Get dataset_uri from DVC for the base artifact dataset_uri = dvc_get_hash(dataset_name) if dataset_uri == "": logger.error(f"[log_label] Error in getting the dvc hash for {dataset_name}, return without logging") @@ -1523,13 +1624,13 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t self.update_existing_artifact(dataset_artifact, dataset_custom_properties) # Prepare label custom properties - custom_props = {} if custom_properties is None else custom_properties + custom_props = {} if custom_properties is None else dict(custom_properties) custom_props["dataset_uri"] = dataset_uri git_repo = git_get_repo() # Check if label artifact already exists existing_artifact = [] - if label_hash and label_hash.strip: + if label_hash and label_hash.strip(): existing_artifact.extend(self.store.get_artifacts_by_uri(label_hash)) url = url + ":" + label_hash @@ -1576,9 +1677,32 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t custom_properties=custom_props, milliseconds_since_epoch=int(time.time() * 1000), ) - custom_props["git_repo"] = git_repo - custom_props["Commit"] = label_hash - + system_props = {"git_repo": git_repo, "Commit": label_hash} + graph_custom_props = {**custom_props, **system_props} + + execution_uuid = self.execution.properties["Execution_uuid"].string_value.split(",")[-1].strip() + if self.custom_store and execution_uuid and artifact: + try: + # Step 1: Get artifact uri and build artifact-level metadata payload. + artifact_uri = str(getattr(artifact, "uri", "")).strip() + label_log_payload = { + "name": name, + "properties": { + "git_repo": git_repo, + "url": url, + }, + "custom_properties": custom_props, + } + # Step 2: Upsert one ExecutionLogs row keyed by (execution_uuid, artifact_uri). + if artifact_uri: + self.custom_store.insert_execution_log( + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + metadata=label_log_payload, + ) + except Exception as exc: + logger.warning("Failed to write label metadata into execution_log: %s", exc) + if self.graph: # directly linked to dataset via create_label_node self.driver.create_label_node( @@ -1589,7 +1713,7 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t self.execution.id, self.parent_context, dataset_uri, # Pass dataset_uri to link label to dataset - custom_props, + graph_custom_props, ) # NOTE: Labels are NOT added to self.input_artifacts to prevent them from being # linked to other artifacts via create_artifact_relationships(). Labels are @@ -1616,7 +1740,6 @@ def log_label(self, url: str, dataset_name: str, custom_properties: t.Optional[t # } # ) - return artifact class DataSlice: diff --git a/cmflib/cmf_federation.py b/cmflib/cmf_federation.py index 9487e8c51..499e96fb8 100644 --- a/cmflib/cmf_federation.py +++ b/cmflib/cmf_federation.py @@ -1,7 +1,93 @@ import json import typing as t +import logging from cmflib.cmfquery import CmfQuery from cmflib.cmf_merger import parse_json_to_mlmd +from cmflib.utils.helper_functions import get_postgres_config + + +logger = logging.getLogger(__name__) + + +def _append_execution_logs_from_pipeline(pipeline_data: dict, cmd: str) -> None: + # Function Name: _append_execution_logs_from_pipeline + # Input: pipeline_data (dict), cmd (str) + # Output: None + # Description: Append ExecutionLogs rows from incoming pipeline JSON payload. + # Step 1: Run only for metadata push command. + # Step 2: Iterate stage -> execution -> event from pipeline payload. + # Step 3: Skip Environment artifacts and rows missing execution_uuid/artifact_uri. + # Step 4: Build metadata_json with name/properties/custom_properties. + # Step 5: Insert one append-only row per artifact event into ExecutionLogs. + # Step 6: Log failures without interrupting the MLMD merge flow. + if cmd != "push": + return + + conn = None + try: + import psycopg # type: ignore + + config = get_postgres_config() + conn = psycopg.connect( + host=config.get("host"), + port=config.get("port"), + user=config.get("user"), + password=config.get("password"), + dbname=config.get("dbname"), + connect_timeout=10, + ) + + with conn.cursor() as cursor: + for stage in pipeline_data.get("stages", []): + for execution in stage.get("executions", []): + execution_uuid = execution.get("properties", {}).get("Execution_uuid", "") + if not execution_uuid: + continue + + for event in execution.get("events", []): + artifact = event.get("artifact", {}) + if artifact.get("type") == "Environment": + continue + + artifact_uri = str(artifact.get("uri", "")).strip() + if not artifact_uri: + continue + + metadata_payload = { + "custom_properties": artifact.get("custom_properties", {}), + "name": artifact.get("name", ""), + "properties": artifact.get("properties", {}), + } + + cursor.execute( + """ + INSERT INTO executionlogs ( + execution_uuid, + artifact_uri, + metadata_json + ) VALUES (%s, %s, %s::jsonb) + """, + ( + execution_uuid, + artifact_uri, + json.dumps(metadata_payload, default=str), + ), + ) + + conn.commit() + except Exception as exc: + logger.warning("Failed to append rows into ExecutionLogs from update_mlmd: %s", exc) + if conn is not None: + try: + conn.rollback() + except Exception: + pass + finally: + if conn is not None: + try: + conn.close() + except Exception: + pass def identify_existing_and_new_executions(query: CmfQuery, pipeline_data: dict, pipeline_name: str) -> t.Tuple[list, list, list, str]: @@ -123,6 +209,10 @@ def update_mlmd(query: CmfQuery, req_info: str, pipeline_name: str, cmd: str, ex if len(pipeline['stages']) == 0 : status="exists" else: + # Side-write: append tracking rows into server ExecutionLogs using parsed JSON payload. + # This does not modify existing MLMD merge behavior. + _append_execution_logs_from_pipeline(pipeline, cmd) + # metadata push → merge client data into server path # metadata pull → merge server data into client path if cmd == "pull": @@ -160,6 +250,10 @@ def update_mlmd(query: CmfQuery, req_info: str, pipeline_name: str, cmd: str, ex if len(pipeline['stages']) == 0 : status="exists" else: + # Side-write: append tracking rows into server ExecutionLogs using parsed JSON payload. + # This does not modify existing MLMD merge behavior. + _append_execution_logs_from_pipeline(pipeline, cmd) + # metadata push → merge client data into server path # metadata pull → merge server data into client path if cmd == "pull": diff --git a/cmflib/cmfquery.py b/cmflib/cmfquery.py index 080dbff3a..f4f9ecac1 100644 --- a/cmflib/cmfquery.py +++ b/cmflib/cmfquery.py @@ -27,6 +27,7 @@ from cmflib.cmf_merger import parse_json_to_mlmd from cmflib.store.postgres import PostgresStore from cmflib.store.sqllite_store import SqlliteStore +from cmflib.store.custom_sqlite_store import CustomSqliteStore from cmflib.utils.helper_functions import get_postgres_config # Constants for filtering artifact and execution types in lineage visualizations @@ -133,6 +134,11 @@ def __init__(self, filepath: str = "mlmd", is_server=False) -> None: else: temp_store = SqlliteStore({"filename": filepath}) self.store = temp_store.connect() + + # Step 1: Initialize CustomSqliteStore to enable reading execution_log during JSON export. + # Step 2: CustomSqliteStore manages DB connection to same mlmd file as MLMD metadata. + # Step 3: Used later in _get_event_attributes() to patch artifacts with execution_log data. + self.custom_store = CustomSqliteStore(filepath) @staticmethod def _copy( @@ -218,6 +224,102 @@ def _as_pandas_df(elements: t.Iterable, transform_fn: t.Callable[[t.Any], pd.Dat df = pd.concat([df, transform_fn(element)], sort=True, ignore_index=True) return df + def _extract_execution_uuids(self, execution_obj) -> t.List[str]: # type: ignore + """Extract execution UUIDs from execution object properties. + + Args: + execution_obj: MLMD execution object with properties containing 'Execution_uuid'. + + Returns: + List of UUIDs as strings. If comma-separated in properties, splits and returns all. + If not found or error, returns empty list. + """ + # Step 1: Validate execution object has properties attribute. + # Step 2: Extract Execution_uuid field (may be comma-separated for multiple UUIDs). + # Step 3: Split by comma and strip whitespace from each UUID. + # Step 4: Return list of UUIDs; return empty list if not found or error. + + try: + # Step 1: Check if execution object has properties. + if not hasattr(execution_obj, 'properties') or not execution_obj.properties: + return [] + + # Step 2: Extract Execution_uuid from properties dict. + if 'Execution_uuid' not in execution_obj.properties: + return [] + + uuid_str = execution_obj.properties['Execution_uuid'].string_value + if not uuid_str: + return [] + + # Step 3: Split by comma if multiple UUIDs, strip whitespace from each. + uuids = [uuid.strip() for uuid in uuid_str.split(",")] + + # Step 4: Filter out empty strings and return non-empty UUIDs. + return [uuid for uuid in uuids if uuid] + except Exception as e: + # Step 5: Return empty list if any error occurs during extraction. + logger.warning(f"Error extracting Execution_uuid from execution: {e}") + return [] + + def _patch_artifact_with_execution_log( + self, artifact_dict: t.Dict[str, t.Any], execution_uuid: str, artifact_uri: t.Any + ) -> None: + """Patch artifact dictionary with execution_log data if entries differ. + + Strategy: Only replace name, properties, custom_properties if execution_log has them + and they differ from current artifact values. Leave unchanged if same or missing. + + Args: + artifact_dict: Artifact metadata dictionary to patch (modified in-place). + execution_uuid: Execution UUID to use as key for lookup. + artifact_uri: Artifact URI to use as key for lookup. + + Returns: + None (modifies artifact_dict in-place). + """ + # Step 1: Query ExecutionLogs for (execution_uuid, artifact_uri) pair. + # Step 2: If found, compare each patchable field (name, properties, custom_properties). + # Step 3: For each field, check if execution_log value differs from artifact value. + # Step 4: If different, replace artifact value with execution_log value. + # Step 5: If same or missing in execution_log, leave artifact unchanged. + + # Step 1: Query custom_store for execution_log metadata. + execution_log_metadata = self.custom_store.get_execution_log(str(execution_uuid), str(artifact_uri)) + + # If no execution_log entry, nothing to patch; return early. + if execution_log_metadata is None: + return + + # Step 2-5: Patchable fields are: name, properties, custom_properties. + # Only these three fields from execution_log are used to patch artifact. + patchable_fields = ["name", "properties", "custom_properties"] + + for field in patchable_fields: + # Step 2: Check if execution_log has this field. + if field not in execution_log_metadata: + # Field not in execution_log; leave artifact unchanged. + continue + + log_value = execution_log_metadata[field] + artifact_value = artifact_dict.get(field) + + # Merge map-like fields so sparse execution_log payloads do not erase + # richer artifact metadata such as labels_uri/custom properties. + if field in {"properties", "custom_properties"} and isinstance(artifact_value, dict) and isinstance(log_value, dict): + merged_value = dict(artifact_value) + merged_value.update(log_value) + if merged_value != artifact_value: + artifact_dict[field] = merged_value + continue + + # Step 3: Compare values to decide if patch is needed. + # Handle case where artifact field might not exist (use None as default). + if log_value != artifact_value: + # Step 4: Values differ; patch artifact with execution_log value. + artifact_dict[field] = log_value + # else: Step 5 - Values same or field missing; leave unchanged. + def _get_pipelines(self, name: t.Optional[str] = None) -> t.List[mlpb.Context]: # type: ignore # Context type not recognized by mypy, using ignore to bypass """Return list of pipelines with the given name. @@ -978,25 +1080,55 @@ def _get_node_attributes(self, _node: t.Union[mlpb.Context, mlpb.Execution, mlpb ) return _attrs - def _get_event_attributes(self, execution_id: int) -> t.List[t.Dict]: + def _get_event_attributes(self, execution_id: int, execution_obj=None) -> t.List[t.Dict]: # type: ignore """ Extract event attributes for a given execution ID. Args: execution_id (int): The ID of the execution for which event attributes are to be extracted. + execution_obj (Optional): Optional MLMD execution object to enable execution_log patching. Returns: List[Dict]: A list of dictionaries, each containing attributes of an event associated with the execution. """ + # Step 1: Retrieve all events for this execution ID. + # Step 2: For each event, extract event attributes and get associated artifact. + # Step 3: Apply artifact patching if execution_obj provided (contains Execution_uuid). + # Step 4: Try each execution UUID (comma-separated) to find matching execution_log entry. + # Step 5: Return list of events with patched artifacts. + events = [] + + # Step 1: Extract execution UUIDs from execution_obj if provided. + execution_uuids = [] + if execution_obj is not None: + execution_uuids = self._extract_execution_uuids(execution_obj) + + # Step 2-5: Process each event. for event in self.store.get_events_by_execution_ids([execution_id]): + # Step 2a: Get event attributes (e.g., type, artifact_id, etc.). event_attrs = self._get_node_attributes(event, {}) + + # Step 2b: Get artifact associated with this event. artifacts = self.store.get_artifacts_by_id([event.artifact_id]) artifact_attrs = self._get_node_attributes( artifacts[0], {"type": self.store.get_artifact_types_by_id([artifacts[0].type_id])[0].name} ) + + # Step 3: If execution_uuids available, try patching artifact with execution_log data. + if execution_uuids and artifact_attrs.get("uri"): + # Step 4: Try each execution UUID (any match from comma-separated list is accepted). + for uuid in execution_uuids: + # Try to patch artifact with this UUID. + self._patch_artifact_with_execution_log(artifact_attrs, uuid, artifact_attrs["uri"]) + # Note: _patch_artifact_with_execution_log is idempotent; calling it multiple times is safe. + # If one UUID matches, the patch is applied; other UUIDs won't match. + # For efficiency, could break after first successful patch, but multiple attempts are safe. + + # Step 5: Add patched artifact to event attributes. event_attrs["artifact"] = artifact_attrs events.append(event_attrs) + return events def _get_execution_attributes(self, stage_id: int, exec_uuid: t.Optional[str] = None, last_sync_time: t.Optional[int] = None) -> t.List[t.Dict]: @@ -1010,26 +1142,42 @@ def _get_execution_attributes(self, stage_id: int, exec_uuid: t.Optional[str] = Returns: List[Dict]: A list of dictionaries, each containing attributes of an execution associated with the stage. """ + # Step 1: Get all executions for this stage, optionally filtered by exec_uuid. + # Step 2: For each execution, extract attributes and nested events/artifacts. + # Step 3: Pass execution object to _get_event_attributes so patching can access Execution_uuid. + # Step 4: Apply last_sync_time filter if provided. + # Step 5: Return list of execution attributes. + executions = [] + + # Step 1-2: Process each execution. for execution in self.get_all_executions_by_stage(stage_id, execution_uuid=exec_uuid): + # Step 3: Extract events by passing execution object to enable execution_log patching. + events = self._get_event_attributes(execution.id, execution) + + # Build execution attributes dict with nested events. exec_attrs = self._get_node_attributes( execution, { "type": self.store.get_execution_types_by_id([execution.type_id])[0].name, "name": execution.name if execution.name != "" else "", - "events": self._get_event_attributes(execution.id), + "events": events, # Use events variable with patched artifacts. }, ) + # Step 4: Filter by last_sync_time if provided. + # If last_sync_time given, only include execution if it was updated after that timestamp. + # If no last_sync_time, include all executions. # what is usual situtaion - #it does not matter if last sync time is given or not we have to add exec_attrs #however if last sync timr is given then we have to check if last_update_time_since_epoch > last_sync_time # last_update_time_since epoch - + if last_sync_time: if exec_attrs["last_update_time_since_epoch"] > last_sync_time: executions.append(exec_attrs) else: + # Step 5: No time filter; add all execution attributes. executions.append(exec_attrs) return executions diff --git a/cmflib/store/custom_sqlite_store.py b/cmflib/store/custom_sqlite_store.py new file mode 100644 index 000000000..3b1c6f93b --- /dev/null +++ b/cmflib/store/custom_sqlite_store.py @@ -0,0 +1,182 @@ +import json +import os +import sqlite3 +import typing as t + + +class CustomSqliteStore: + """Store execution-level artifact metadata in SQLite.""" + + def __init__(self, db_path: str): + # Function Name: __init__ + # Input: db_path (str) + # Output: None + # Description: Resolve DB path and initialize execution_log table. + # Step 1: Normalize incoming DB path. + # Step 2: Create DB/table if missing. + self.db_path = self._resolve_db_path(db_path) + self._init_db() + + @staticmethod + def _resolve_db_path(path_hint: str) -> str: + # Function Name: _resolve_db_path + # Input: path_hint (str) + # Output: str (resolved DB path) + # Description: Determine final DB path based on input hint or defaults. + # Step 1: Trim whitespace from input path hint. + raw_path = (path_hint or "").strip() + + # Step 2: If no path provided, default to current working directory. + if not raw_path: + raw_path = os.getcwd() + + # Step 3: Normalize path to remove redundant separators and up-level references. + normalized_path = os.path.normpath(raw_path) + + # Step 4: If the path ends with "mlmd", treat it as the final DB path. + if os.path.basename(normalized_path) == "mlmd": + return normalized_path + + # Step 5: Otherwise, treat input as a directory path and place DB at /mlmd. + return os.path.join(normalized_path, "mlmd") + + def _connect(self) -> sqlite3.Connection: + # Function Name: _connect + # Input: None + # Output: sqlite3.Connection + # Description: Create a resilient SQLite connection for shared-file usage. + # Step 1: Open connection with a longer timeout. + # Step 2: Configure journaling and sync pragmas for safer writes. + # Step 3: Return configured connection. + # Use a longer timeout to avoid lock-contention failures when MLMD writes to the same DB. + conn = sqlite3.connect(self.db_path, timeout=15) + # Keep classic journaling when sharing one DB file with MLMD. + + # What each line does: + + # 1. `PRAGMA journal_mode=DELETE;` + # - Uses classic rollback journal mode. + # - Reason for change: your previous mode was WAL, and WAL can be fragile with mixed tooling/writers (MLMD writes, custom sqlite writes, VS Code viewer reads). + # - In shared single-file usage, `DELETE` is simpler and usually less likely to produce weird sidecar-file issues (`mlmd-wal`, `mlmd-shm`) that lead to malformed/open errors in tools. + + # 2. `PRAGMA busy_timeout=15000;` + # - Wait up to 15 seconds if DB is locked instead of failing immediately. + # - Reason: MLMD and custom writes can briefly overlap. Without timeout, you get fast lock errors and possibly incomplete operation sequences. + # - This improves stability under lock contention. + + # 3. `PRAGMA synchronous=FULL;` + # - Forces SQLite to fully flush writes to disk before commit is considered done. + # - Reason: strongest durability, reduces risk of corruption if process/system crashes during writes. + # - Tradeoff: slightly slower writes, but safer. + + # Why this set specifically: + # - You asked to keep a single DB (no separate custom DB). + # - Single-DB multi-writer scenarios need safety over speed. + # - These settings prioritize consistency/durability and reduce corruption probability. + + conn.execute("PRAGMA journal_mode=DELETE;") + conn.execute("PRAGMA busy_timeout=15000;") + conn.execute("PRAGMA synchronous=FULL;") + return conn + + def _init_db(self) -> None: + # Function Name: _init_db + # Input: None + # Output: None + # Description: Ensure ExecutionLogs table exists with expected key shape. + # Step 1: Create parent directory when path contains one. + # Step 2: Create ExecutionLogs table if missing. + # Step 3: Commit schema setup. + db_dir = os.path.dirname(self.db_path) + if db_dir: + os.makedirs(db_dir, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS ExecutionLogs ( + execution_uuid TEXT NOT NULL, + artifact_uri TEXT NOT NULL, + metadata_json TEXT NOT NULL, + PRIMARY KEY (execution_uuid, artifact_uri) + ) + """ + ) + conn.commit() + + def insert_execution_log( + self, + execution_uuid: str, + artifact_uri: str, + metadata: t.Optional[t.Dict[str, t.Any]] = None, + ) -> None: + # Function Name: insert_execution_log + # Input: execution_uuid (str), artifact_uri (str), metadata (dict | None) + # Output: None + # Description: Upsert one row per (execution_uuid, artifact_uri) with JSON metadata. + # Step 1: Validate required keys. + # Step 2: Serialize metadata to JSON string. + # Step 3: Insert new row or update existing row for same composite key. + # Step 4: Commit transaction. + if not execution_uuid or not artifact_uri: + return + payload = json.dumps(metadata or {}, default=str, sort_keys=True) + with self._connect() as conn: + conn.execute( + """ + INSERT INTO ExecutionLogs ( + execution_uuid, + artifact_uri, + metadata_json + ) VALUES (?, ?, ?) + ON CONFLICT(execution_uuid, artifact_uri) + DO UPDATE SET metadata_json = excluded.metadata_json + """, + ( + execution_uuid, + artifact_uri, + payload, + ), + ) + conn.commit() + + def get_execution_log(self, execution_uuid: str, artifact_uri: str) -> t.Optional[t.Dict[str, t.Any]]: + # Function Name: get_execution_log + # Input: execution_uuid (str), artifact_uri (str) + # Output: Dict or None + # Description: Retrieve metadata from ExecutionLogs for (execution_uuid, artifact_uri) pair. + # Step 1: Validate input parameters are not empty. + # Step 2: Query ExecutionLogs table using composite key. + # Step 3: If row found, parse metadata_json and return as dict. + # Step 4: If row not found or error occurs, return None. + # Step 5: Log errors if JSON parsing fails. + + # Step 1: Skip retrieval if input is invalid. + if not execution_uuid or not artifact_uri: + return None + + try: + # Step 2: Open connection and query ExecutionLogs by composite key (execution_uuid, artifact_uri). + with self._connect() as conn: + cursor = conn.execute( + """ + SELECT metadata_json FROM ExecutionLogs + WHERE execution_uuid = ? AND artifact_uri = ? + """, + (execution_uuid, artifact_uri), + ) + row = cursor.fetchone() + + # Step 3: If row found, deserialize metadata_json from JSON string to dict. + if row: + return json.loads(row[0]) + + # Step 4: Row not found; return None to indicate no custom metadata for this pair. + return None + except json.JSONDecodeError as e: + # Step 5a: Log error if JSON parsing fails (data corruption scenario). + print(f"Error parsing JSON for ({execution_uuid}, {artifact_uri}): {e}") + return None + except Exception as e: + # Step 5b: Catch other exceptions (DB errors) and return None; caller will use original artifact data. + print(f"Error querying ExecutionLogs for ({execution_uuid}, {artifact_uri}): {e}") + return None \ No newline at end of file diff --git a/server/app/db/dbmodels.py b/server/app/db/dbmodels.py index 7b4d502de..edff79188 100644 --- a/server/app/db/dbmodels.py +++ b/server/app/db/dbmodels.py @@ -12,8 +12,10 @@ Index, UniqueConstraint, MetaData, - SmallInteger + SmallInteger, + text ) +from sqlalchemy.dialects.postgresql import JSONB metadata = MetaData() @@ -205,6 +207,18 @@ ) +executionlogs = Table( + "executionlogs", metadata, + Column("execution_uuid", Text, nullable=False), + Column("artifact_uri", Text, nullable=False), + Column("metadata_json", JSONB, nullable=False, server_default=text("'{}'::jsonb")), + + # Indexes + Index("idx_executionlogs_execution_uuid", "execution_uuid"), + Index("idx_executionlogs_artifact_uri", "artifact_uri"), + Index("idx_executionlogs_exec_uri", "execution_uuid", "artifact_uri"), + + # Schedules for periodic syncs scheduled_syncs = Table( "scheduled_syncs", metadata, diff --git a/server/app/db/dbqueries.py b/server/app/db/dbqueries.py index b526ab14e..72d8b525a 100644 --- a/server/app/db/dbqueries.py +++ b/server/app/db/dbqueries.py @@ -14,6 +14,7 @@ executionproperty, event, registered_servers, + executionlogs, scheduled_syncs, sync_logs ) @@ -676,6 +677,19 @@ async def fetch_artifacts_by_stage( effective_page_size = record_per_page if record_per_page is not None else page_size + # Step 0: Aggregate execution_count and distinct names per artifact URI from ExecutionLogs. + execution_log_agg_cte = ( + select( + executionlogs.c.artifact_uri, + func.count(distinct(executionlogs.c.execution_uuid)).label("execution_count"), + func.json_agg( + distinct(func.cast(executionlogs.c.metadata_json[text("'name'")], String)) + ).label("execution_names") + ) + .group_by(executionlogs.c.artifact_uri) + .subquery("execution_log_agg_cte") + ) + # Step 1: Aggregate artifact properties into JSON based on artifact id. artifact_properties_agg_cte = ( select( @@ -705,6 +719,7 @@ async def fetch_artifacts_by_stage( select( artifact.c.id.label("artifact_id"), artifact.c.name, + artifact.c.uri, artifact.c.create_time_since_epoch ) .join( @@ -755,8 +770,11 @@ async def fetch_artifacts_by_stage( select( artifact_type_cte.c.artifact_id, artifact_type_cte.c.name, + artifact_type_cte.c.uri, artifact_type_cte.c.create_time_since_epoch, artifact_properties_agg_cte.c.artifact_properties, + func.coalesce(execution_log_agg_cte.c.execution_count, 0).label("execution_count"), + execution_log_agg_cte.c.execution_names, func.count().over().label("total_records") ) .select_from(artifact_type_cte) @@ -764,6 +782,10 @@ async def fetch_artifacts_by_stage( artifact_properties_agg_cte, artifact_type_cte.c.artifact_id == artifact_properties_agg_cte.c.artifact_id ) + .outerjoin( + execution_log_agg_cte, + artifact_type_cte.c.uri == execution_log_agg_cte.c.artifact_uri + ) ) # Apply search filter if provided @@ -796,6 +818,83 @@ async def fetch_artifacts_by_stage( } +async def fetch_execution_uuids_by_artifact_uri( + db: AsyncSession, + pipeline_name: str, + artifact_uri: str, +): + """Return execution UUIDs and names from ExecutionLogs for a given artifact URI scoped to pipeline.""" + if not artifact_uri: + return [] + + _, pipeline_execution_ids = _get_pipeline_execution_ids_subquery(pipeline_name) + + pipeline_execution_uuids = ( + select(distinct(executionproperty.c.string_value).label("execution_uuid")) + .where(executionproperty.c.name == "Execution_uuid") + .where(executionproperty.c.string_value.isnot(None)) + .where(executionproperty.c.execution_id.in_(select(pipeline_execution_ids.c.execution_id))) + .subquery("pipeline_execution_uuids") + ) + + query = ( + select( + executionlogs.c.execution_uuid.label("execution_uuid"), + func.max(func.cast(executionlogs.c.metadata_json[text("'name'")], String)).label("name"), + ) + .where(executionlogs.c.artifact_uri == artifact_uri) + .where(executionlogs.c.execution_uuid.in_(select(pipeline_execution_uuids.c.execution_uuid))) + .group_by(executionlogs.c.execution_uuid) + .order_by(executionlogs.c.execution_uuid.desc()) + ) + + result = await db.execute(query) + rows = result.mappings().all() + return [ + { + "execution_uuid": row["execution_uuid"], + "name": row.get("name"), + } + for row in rows + ] + + +async def fetch_execution_log_metadata( + db: AsyncSession, + pipeline_name: str, + execution_uuid: str, + artifact_uri: str, +): + """Return metadata_json from ExecutionLogs for one (execution_uuid, artifact_uri) scoped to pipeline.""" + if not execution_uuid or not artifact_uri: + return None + + _, pipeline_execution_ids = _get_pipeline_execution_ids_subquery(pipeline_name) + + pipeline_execution_uuids = ( + select(distinct(executionproperty.c.string_value).label("execution_uuid")) + .where(executionproperty.c.name == "Execution_uuid") + .where(executionproperty.c.string_value.isnot(None)) + .where(executionproperty.c.execution_id.in_(select(pipeline_execution_ids.c.execution_id))) + .subquery("pipeline_execution_uuids") + ) + + query = ( + select(executionlogs.c.metadata_json) + .where(executionlogs.c.execution_uuid == execution_uuid) + .where(executionlogs.c.artifact_uri == artifact_uri) + .where(executionlogs.c.execution_uuid.in_(select(pipeline_execution_uuids.c.execution_uuid))) + .limit(1) + ) + + result = await db.execute(query) + row = result.first() + if not row: + return None + + return row[0] + + async def fetch_artifact_types_by_stage( db: AsyncSession, pipeline_name: str, diff --git a/server/app/main.py b/server/app/main.py index db2c6023c..ef85ab4e1 100644 --- a/server/app/main.py +++ b/server/app/main.py @@ -36,6 +36,8 @@ fetch_executions_by_stage, fetch_artifacts_by_stage, fetch_artifact_types_by_stage, + fetch_execution_uuids_by_artifact_uri, + fetch_execution_log_metadata, register_server_details, get_registered_server_details, get_sync_status, @@ -500,6 +502,47 @@ async def get_artifacts_by_stage( ) +@app.get("/artifact-executions/{pipeline_name}") +async def get_artifact_execution_uuids( + pipeline_name: str, + artifact_uri: str = Query(..., description="Artifact URI"), + db: AsyncSession = Depends(get_db) +): + """Return execution UUIDs and names for a selected artifact URI from ExecutionLogs.""" + executions = await fetch_execution_uuids_by_artifact_uri( + db=db, + pipeline_name=pipeline_name, + artifact_uri=artifact_uri, + ) + return { + "artifact_uri": artifact_uri, + "executions": executions, + "execution_uuids": [item.get("execution_uuid") for item in executions if item.get("execution_uuid")], + } + + +@app.get("/artifact-execution-metadata/{pipeline_name}") +async def get_artifact_execution_metadata( + pipeline_name: str, + execution_uuid: str = Query(..., description="Execution UUID"), + artifact_uri: str = Query(..., description="Artifact URI"), + db: AsyncSession = Depends(get_db) +): + """Return metadata_json for a selected (execution_uuid, artifact_uri) from ExecutionLogs.""" + metadata = await fetch_execution_log_metadata( + db=db, + pipeline_name=pipeline_name, + execution_uuid=execution_uuid, + artifact_uri=artifact_uri, + ) + + return { + "execution_uuid": execution_uuid, + "artifact_uri": artifact_uri, + "metadata": metadata or {}, + } + + @app.get("/execution-lineage/tangled-tree/{uuid}/{pipeline_name}") async def execution_lineage(request: Request, uuid: str, pipeline_name: str): ''' diff --git a/ui/src/client.js b/ui/src/client.js index f1b6e6484..6a484db53 100644 --- a/ui/src/client.js +++ b/ui/src/client.js @@ -328,6 +328,31 @@ class FastAPIClient { }); } + async getArtifactExecutions(pipelineName, artifactUri) { + return this.apiClient + .get(`/artifact-executions/${pipelineName}`, { + params: { + artifact_uri: artifactUri, + }, + }) + .then(({ data }) => { + return data; + }); + } + + async getArtifactExecutionMetadata(pipelineName, executionUuid, artifactUri) { + return this.apiClient + .get(`/artifact-execution-metadata/${pipelineName}`, { + params: { + execution_uuid: executionUuid, + artifact_uri: artifactUri, + }, + }) + .then(({ data }) => { + return data; + }); + } + } diff --git a/ui/src/components/ArtifactCardGrid/index.jsx b/ui/src/components/ArtifactCardGrid/index.jsx index 4ab56242a..c0b8d7f8c 100644 --- a/ui/src/components/ArtifactCardGrid/index.jsx +++ b/ui/src/components/ArtifactCardGrid/index.jsx @@ -32,6 +32,7 @@ const ArtifactCardGrid = ({ isSplitView = false, selectedItems = [], onToggleItem, + selectedArtifactId = null, }) => { const [expandedCard, setExpandedCard] = useState(null); const [showModelPopup, setShowModelPopup] = useState(false); @@ -76,6 +77,16 @@ const ArtifactCardGrid = ({ } }; + const normalizeStepMetricsName = (value) => { + const text = String(value || ""); + return text.toLowerCase().includes("training_metrics") ? "training_metrics" : text; + }; + + const getCardDisplayName = (artifact) => { + const baseValue = artifact?.uri || artifact?.name || "N/A"; + return normalizeStepMetricsName(baseValue); + }; + const renderLabels = (artifact) => { const labelsUri = getPropertyValue(artifact.artifact_properties, "labels_uri"); @@ -92,7 +103,7 @@ const ArtifactCardGrid = ({ .map((label_name, idx) => ( { e.preventDefault(); e.stopPropagation(); @@ -102,7 +113,7 @@ const ArtifactCardGrid = ({ }} title={label_name} > - + @@ -149,14 +160,16 @@ const ArtifactCardGrid = ({ {artifacts.map((artifact, index) => (
a.artifact_id === artifact.artifact_id) - ? 'border-teal-500 shadow-lg' - : 'border-gray-300 hover:border-teal-500 hover:shadow-lg' + className={`rounded-lg border-2 ${selectedArtifactId === artifact.artifact_id + ? 'bg-cyan-50 border-cyan-500 shadow-lg' + : selectedItems.some(a => a.artifact_id === artifact.artifact_id) + ? 'bg-teal-50 border-teal-500 shadow-lg' + : 'bg-white border-gray-300 hover:bg-teal-50 hover:border-teal-500 hover:shadow-lg' } transition-all duration-200 overflow-hidden ${onArtifactClick ? 'cursor-pointer' : ''}`} onClick={() => onArtifactClick && onArtifactClick(artifact)} > {/* Card Header */} -
+
@@ -171,7 +184,8 @@ const ArtifactCardGrid = ({ {artifactType}
-

+

+ URI: {artifactType === "Label" && onLabelClick ? ( - + ) : ( - + )}

@@ -225,60 +239,60 @@ const ArtifactCardGrid = ({ Created: - + {formatDate(artifact.create_time_since_epoch)}
- {/* URI */} - {artifact.uri && artifact.uri !== "N/A" && ( -
- - - -
- URI: - - - -
-
- )} - - {/* Git Info */} - {(getPropertyValue(artifact.artifact_properties, "git_repo") !== "N/A" || - getPropertyValue(artifact.artifact_properties, "Commit") !== "N/A") && ( -
- {getPropertyValue(artifact.artifact_properties, "git_repo") !== "N/A" && ( -
- - - - - - -
- )} - {getPropertyValue(artifact.artifact_properties, "Commit") !== "N/A" && ( -
- - - - - + {/* Artifact Names from ExecutionLogs */} + {(() => { + const rawNames = artifact.execution_names; + const stripQuotes = n => typeof n === 'string' ? n.replace(/^"|"$/g, '') : n; + const names = Array.isArray(rawNames) + ? rawNames.map(stripQuotes).filter(n => n && n !== 'null') + : (typeof rawNames === 'string' + ? (() => { try { const p = JSON.parse(rawNames); return Array.isArray(p) ? p.map(stripQuotes).filter(n => n && n !== 'null') : []; } catch { return []; } })() + : []); + const uniqueDisplayNames = [...new Set(names.map((n) => normalizeStepMetricsName(n)))]; + if (uniqueDisplayNames.length === 0) return null; + return ( +
+
+ {uniqueDisplayNames.map((n, i) => ( + + -
- )} + ))} +
- )} + ); + })()} {/* Labels for Dataset */} {artifactType === "Dataset" && ( -
- Labels: +
+ Labels: {renderLabels(artifact)}
)} + {/* Execution count badge */} + {artifact.execution_count != null && Number(artifact.execution_count) > 0 && ( +
+
+ + + +
+ + Linked in {" "} + + {artifact.execution_count} + {" "} + {Number(artifact.execution_count) === 1 ? "execution" : "executions"} + +
+ )} + {/* Model Card Button */} {artifactType === "Model" && (
- - {/* Card Footer - Expandable Properties */} -
- - - {expandedCard === index && ( -
-
- {artifact.artifact_properties?.map((property, idx) => ( -
-
- -
-
- -
-
- ))} -
-
- )} -
))}
diff --git a/ui/src/components/DetailDrawer/index.jsx b/ui/src/components/DetailDrawer/index.jsx index 2ad7e181e..fc8ec36ae 100644 --- a/ui/src/components/DetailDrawer/index.jsx +++ b/ui/src/components/DetailDrawer/index.jsx @@ -14,7 +14,7 @@ * limitations under the License. ***/ -import React from "react"; +import React, { useState } from "react"; /** * Generic reusable detail drawer that slides in from the right. @@ -22,11 +22,21 @@ import React from "react"; * Props: * - title {string} — Drawer heading (e.g. "Execution Details") * - subtitle {node} — Sub-heading node rendered below the title - * - summaryFields {Array} — [{ label, value, color }] key-info chips * - allProperties {Array} — [{ name, value }] full property list * - onClose {function} — Called when overlay or ✕ is clicked + * - children {node} — Optional custom content rendered below properties + * - showAllProperties {boolean} — Toggle All Properties section visibility */ -const DetailDrawer = ({ title, subtitle, summaryFields = [], allProperties = [], onClose }) => { +const DetailDrawer = ({ + title, + subtitle, + allProperties = [], + onClose, + children, + showAllProperties = true, +}) => { + const [isFullscreen, setIsFullscreen] = useState(false); + if (!onClose) return null; return ( @@ -38,62 +48,78 @@ const DetailDrawer = ({ title, subtitle, summaryFields = [], allProperties = [], /> {/* Drawer panel */} -
+
{/* Header */}

{title}

{subtitle && ( -

{subtitle}

+
{subtitle}
)}
- +
+ + +
{/* Body */}
- {/* Summary chips */} - {summaryFields.length > 0 && ( -
- {summaryFields.map(({ label, value, color }) => - value && value !== "N/A" ? ( -
-
{label}
-
{value}
+ {showAllProperties && ( + <> + {/* Divider + section title */} +
+

Properties

+
+ + {/* All properties */} +
+ {allProperties.map((prop, idx) => ( +
+
{prop.name}
+
{prop.value}
- ) : null - )} -
+ ))} + {allProperties.length === 0 && ( +

No properties available

+ )} +
+ )} - {/* Divider + section title */} -
-

All Properties

-
- - {/* All properties */} -
- {allProperties.map((prop, idx) => ( -
-
{prop.name}
-
{prop.value}
-
- ))} - {allProperties.length === 0 && ( -

No properties available

- )} -
+ {children && ( +
+ {children} +
+ )}
diff --git a/ui/src/components/ExecutionCardGrid/index.jsx b/ui/src/components/ExecutionCardGrid/index.jsx index c617f615a..16c9ce75c 100644 --- a/ui/src/components/ExecutionCardGrid/index.jsx +++ b/ui/src/components/ExecutionCardGrid/index.jsx @@ -22,7 +22,7 @@ import PythonEnvPopup from "../PythonEnvPopup"; const client = new FastAPIClient(config); -const ExecutionCard = ({ execution, filterValue, onCardClick, isSelected = false, onToggle }) => { +const ExecutionCard = ({ execution, filterValue, onCardClick, isSelected = false, isActive = false, onToggle }) => { const [showPythonPopup, setShowPythonPopup] = useState(false); const [pythonEnvData, setPythonEnvData] = useState(""); const [expandedProperties, setExpandedProperties] = useState(false); @@ -88,23 +88,21 @@ const ExecutionCard = ({ execution, filterValue, onCardClick, isSelected = false return ( <>
onCardClick && onCardClick(execution)} > {/* Card Header */} -
+
- - + +
@@ -123,7 +121,7 @@ const ExecutionCard = ({ execution, filterValue, onCardClick, isSelected = false - + {createdAt}
)}

@@ -197,48 +195,6 @@ const ExecutionCard = ({ execution, filterValue, onCardClick, isSelected = false )}

- - {/* Card Footer - Expandable all properties */} -
- - {expandedProperties && ( -
-
- {allProps.map((prop, idx) => ( -
-
- -
-
- -
-
- ))} -
-
- )} -
{showPythonPopup && ( diff --git a/ui/src/pages/artifacts_postgres_grid/index.jsx b/ui/src/pages/artifacts_postgres_grid/index.jsx index 0b278a9ca..971b5a729 100644 --- a/ui/src/pages/artifacts_postgres_grid/index.jsx +++ b/ui/src/pages/artifacts_postgres_grid/index.jsx @@ -31,6 +31,72 @@ import Papa from "papaparse"; const client = new FastAPIClient(config); +const getArtifactPropertiesArray = (artifact) => { + if (!artifact) return []; + + if (Array.isArray(artifact.artifact_properties)) { + return artifact.artifact_properties; + } + + try { + return JSON.parse(artifact.artifact_properties || "[]"); + } catch { + return []; + } +}; + +const getArtifactPropertyValue = (properties, propertyName, fallback = "N/A") => { + console.log("Getting property value for", propertyName, "from properties:", properties); + const matchedValues = properties + .filter((property) => property.name === propertyName) + .map((property) => property.value) + .filter(Boolean); + + const uniqueValues = [...new Set(matchedValues)]; + return uniqueValues.length > 0 ? uniqueValues.join(", ") : fallback; +}; + +const truncateCommit = (commit) => { + if (!commit || commit === "N/A") return commit; + return commit.length > 12 ? commit.slice(0, 12) + "…" : commit; +}; + +const formatDrawerDate = (timestamp) => { + if (!timestamp || timestamp === "N/A") return "N/A"; + try { + return new Date(parseInt(timestamp)).toLocaleString(); + } catch { + return timestamp; + } +}; + +const normalizeStepMetricsName = (value) => { + const text = String(value || ""); + return text.toLowerCase().includes("training_metrics") ? "training_metrics" : text; +}; + +const sanitizeExecutionName = (value) => { + if (value == null) return null; + return String(value).replace(/^"+|"+$/g, ""); +}; + +const normalizeExecutionRecords = (response) => { + const executions = response?.executions; + if (Array.isArray(executions) && executions.length > 0) { + return executions + .map((entry) => ({ + execution_uuid: entry?.execution_uuid || null, + name: sanitizeExecutionName(entry?.name), + })) + .filter((entry) => Boolean(entry.execution_uuid)); + } + + const executionUuids = response?.execution_uuids || []; + return executionUuids + .map((uuid) => ({ execution_uuid: uuid, name: null })) + .filter((entry) => Boolean(entry.execution_uuid)); +}; + const ArtifactsPostgres = () => { const [searchParams] = useSearchParams(); const [selectedPipeline, setSelectedPipeline] = useState(null); @@ -49,6 +115,11 @@ const ArtifactsPostgres = () => { // Artifact detail drawer state const [selectedArtifact, setSelectedArtifact] = useState(null); + const [artifactExecutionUuids, setArtifactExecutionUuids] = useState([]); + const [selectedExecutionUuid, setSelectedExecutionUuid] = useState(null); + const [selectedExecutionMetadata, setSelectedExecutionMetadata] = useState(null); + const [executionListLoading, setExecutionListLoading] = useState(false); + const [executionMetadataLoading, setExecutionMetadataLoading] = useState(false); // Label content panel states const [selectedTableLabel, setSelectedTableLabel] = useState(null); @@ -181,10 +252,55 @@ const ArtifactsPostgres = () => { setSelectedArtifacts([]); }; - const handleArtifactCardClick = (artifact) => { + const handleArtifactCardClick = async (artifact) => { setSelectedArtifact(artifact); + setArtifactExecutionUuids([]); + setSelectedExecutionUuid(null); + setSelectedExecutionMetadata(null); + + if (!artifact?.uri || !selectedPipeline) return; + + setExecutionListLoading(true); + try { + const response = await client.getArtifactExecutions(selectedPipeline, artifact.uri); + const executionRecords = normalizeExecutionRecords(response); + setArtifactExecutionUuids(executionRecords); + setSelectedExecutionUuid(executionRecords.length > 0 ? executionRecords[0].execution_uuid : null); + } catch (error) { + console.error("Error fetching execution UUIDs for artifact:", error); + setArtifactExecutionUuids([]); + setSelectedExecutionUuid(null); + } finally { + setExecutionListLoading(false); + } }; + useEffect(() => { + const fetchExecutionMetadata = async () => { + if (!selectedPipeline || !selectedArtifact?.uri || !selectedExecutionUuid) { + setSelectedExecutionMetadata(null); + return; + } + + setExecutionMetadataLoading(true); + try { + const response = await client.getArtifactExecutionMetadata( + selectedPipeline, + selectedExecutionUuid, + selectedArtifact.uri, + ); + setSelectedExecutionMetadata(response?.metadata || {}); + } catch (error) { + console.error("Error fetching execution metadata:", error); + setSelectedExecutionMetadata(null); + } finally { + setExecutionMetadataLoading(false); + } + }; + + fetchExecutionMetadata(); + }, [selectedPipeline, selectedArtifact, selectedExecutionUuid]); + const handleFilter = (value) => { setFilter(value); setActivePage(1); @@ -251,39 +367,21 @@ const ArtifactsPostgres = () => { }; const artifactDetailProperties = selectedArtifact - ? (Array.isArray(selectedArtifact.artifact_properties) - ? selectedArtifact.artifact_properties - : (() => { - try { - return JSON.parse(selectedArtifact.artifact_properties || "[]"); - } catch { - return []; - } - })()) + ? getArtifactPropertiesArray(selectedArtifact) : []; - const getArtifactDetailProperty = (name) => { - const matchedValues = artifactDetailProperties - .filter((property) => property.name === name) - .map((property) => property.value); - return matchedValues.length > 0 ? matchedValues.join(", ") : "N/A"; - }; - - const formatArtifactDetailDate = (timestamp) => { - if (!timestamp || timestamp === "N/A") return "N/A"; - try { - return new Date(parseInt(timestamp)).toLocaleString(); - } catch { - return timestamp; - } - }; + const artifactCommitValue = selectedArtifact?.uri || getArtifactPropertyValue(artifactDetailProperties, "Commit"); + const selectedExecutionRecord = artifactExecutionUuids.find((entry) => entry.execution_uuid === selectedExecutionUuid); + const selectedExecutionDisplayName = normalizeStepMetricsName( + sanitizeExecutionName(selectedExecutionRecord?.name || selectedExecutionMetadata?.name) || "N/A", + ); + // Summary fields: immutable artifact metadata (Type, Created At, URI as unique identifier) const artifactDetailSummaryFields = selectedArtifact ? [ { label: "Type", value: selectedArtifactType, color: "teal" }, - { label: "URI", value: selectedArtifact.uri, color: "blue" }, - { label: "Created", value: formatArtifactDetailDate(selectedArtifact.create_time_since_epoch), color: "indigo" }, - { label: "Git Repo", value: getArtifactDetailProperty("git_repo"), color: "purple" }, - { label: "Commit", value: getArtifactDetailProperty("Commit"), color: "green" }, + { label: "Created", value: formatDrawerDate(selectedArtifact.create_time_since_epoch), color: "indigo" }, + { label: "URI (Unique ID)", value: selectedArtifact.uri, color: "blue" }, + { label: "Commit", value: artifactCommitValue, color: "violet" }, ] : []; return ( @@ -434,6 +532,7 @@ const ArtifactsPostgres = () => { onLabelClick={handleLabelClick} onArtifactClick={handleArtifactCardClick} isSplitView={true} selectedItems={selectedArtifacts} + selectedArtifactId={selectedArtifact?.artifact_id} onToggleItem={handleToggleArtifact} /> { filterValue={filter} onArtifactClick={handleArtifactCardClick} selectedItems={selectedArtifacts} + selectedArtifactId={selectedArtifact?.artifact_id} onToggleItem={handleToggleArtifact} /> { {selectedArtifact && ( ID: {selectedArtifact.artifact_id || "—"}} - summaryFields={artifactDetailSummaryFields} + subtitle={ +
+ ID: + {selectedArtifact.artifact_id || "—"} + Commit: + {artifactCommitValue} +
+ } allProperties={artifactDetailProperties} - onClose={() => setSelectedArtifact(null)} - /> + showAllProperties={false} + onClose={() => { + setSelectedArtifact(null); + setArtifactExecutionUuids([]); + setSelectedExecutionUuid(null); + setSelectedExecutionMetadata(null); + }} + > +
+
+ {/* Execution UUID List */} +
+
+

Executions (Name + UUID)

+ {artifactExecutionUuids.length} +
+
+ {executionListLoading ? ( +

Loading execution UUIDs...

+ ) : artifactExecutionUuids.length > 0 ? ( + artifactExecutionUuids.map((execution, idx) => { + const executionUuid = execution.execution_uuid; + const executionName = normalizeStepMetricsName( + sanitizeExecutionName(execution.name) || "N/A", + ); + const isActive = selectedExecutionUuid === executionUuid; + return ( + + ); + }) + ) : ( +

No execution versions found

+ )} +
+
+ + {/* Metadata Snapshot Panel */} +
+ {executionMetadataLoading ? ( +
+

Loading metadata...

+
+ ) : selectedExecutionUuid ? ( +
+
+

Execution UUID

+

+ {selectedExecutionUuid} +

+
+ + {/* Name Field */} + {selectedExecutionMetadata?.name && ( +
+

Name

+
+
{selectedExecutionDisplayName}
+
+
+ )} + + {/* Properties Section */} + {selectedExecutionMetadata?.properties && Object.keys(selectedExecutionMetadata.properties).length > 0 && ( +
+

Properties

+
+ {Object.entries(selectedExecutionMetadata.properties).map(([key, value]) => ( +
+
{key}
+
{String(value)}
+
+ ))} +
+
+ + )} + + {/* Custom Properties Section */} + {selectedExecutionMetadata?.custom_properties && Object.keys(selectedExecutionMetadata.custom_properties).length > 0 && ( +
+

Custom Properties

+
+ {Object.entries(selectedExecutionMetadata.custom_properties).map(([key, value]) => ( +
+
{key}
+
{String(value)}
+
+ ))} +
+
+ )} + + {!selectedExecutionMetadata && ( +
No metadata found for selected execution.
+ )} +
+ ) : ( +
+

Select an execution version to view its metadata

+
+ )} +
+
+
+ )} {/* Compare Modal */} diff --git a/ui/src/pages/executions_postgres_grid/index.jsx b/ui/src/pages/executions_postgres_grid/index.jsx index d28ec3405..feeb72389 100644 --- a/ui/src/pages/executions_postgres_grid/index.jsx +++ b/ui/src/pages/executions_postgres_grid/index.jsx @@ -310,6 +310,7 @@ const ExecutionsPostgresGrid = () => { filterValue={filter} onCardClick={setSelectedExecution} isSelected={selectedExecutions.some(e => e.execution_id === execution.execution_id)} + isActive={selectedExecution?.execution_id === execution.execution_id} onToggle={handleToggleExecution} /> ))} @@ -366,7 +367,6 @@ const ExecutionsPostgresGrid = () => { ID: {selectedExecution.execution_id || "—"}} - summaryFields={executionDetailSummaryFields} allProperties={executionDetailProperties} onClose={() => setSelectedExecution(null)} />