Sync with upstream metrics emit in modin-project/modin#7550

Author: sfc-gh-jkew

setup.py

@@ -44,7 +44,7 @@
     *PANDAS_REQUIREMENTS,
     ### For private customer releases 
     #"modin @ git+https://github.com/sfc-gh-joshi/modin.git@joshi/hybrid-do-not-release",  # TODO point at main
-    "modin @ git+https://github.com/modin-project/modin.git@hybrid-client-alpha",
+    "modin @ git+https://github.com/modin-project/modin.git@main",
     "tqdm",
 ]
 DEVELOPMENT_REQUIREMENTS = [

src/snowflake/snowpark/modin/plugin/_internal/telemetry.py

@@ -625,10 +625,8 @@ def snowpark_pandas_api_watcher(api_name: str, _time: Union[int, float]) -> None
 )
 
 @cached(cache={})
-def get_user_source_location(mode, group) -> str:
-    
+def get_user_source_location(group) -> str:
     import inspect
-
     stack = inspect.stack()
     frame_before_snowpandas = None
     location = "<unknown>"
@@ -644,81 +642,43 @@ def get_user_source_location(mode, group) -> str:
         and frame_before_snowpandas.code_context is not None
     ):
         location = frame_before_snowpandas.code_context[0].replace("\n", "")
-    return {'mode': mode, 'group': group, 'location': location }
+    return { 'group': group, 'source': location }
+
+
 
 def get_hybrid_switch_log():
     global hybrid_switch_log
     return hybrid_switch_log.copy()
-
-def add_to_hybrid_switch_log(metrics: dict):
-    global hybrid_switch_log
-    try:
-        mode = metrics['mode']
-        source = get_user_source_location(mode, metrics['group'])['location']
-        if len(source) > 40:
-            source = source[0:17] + "..." + source[-20:-1] + source[-1]
-        hybrid_switch_log = native_pd.concat([hybrid_switch_log, 
-                                                native_pd.DataFrame({'source': [source],
-                                                                   'mode': [metrics['mode']],
-                                                                   'group': [metrics['group']],
-                                                                   'metric': [metrics['metric']],
-                                                                   'submetric': [metrics['submetric'] or None],
-                                                                   'value': [metrics['value']],
-                                                                   'from': [metrics['from'] if 'from' in metrics else None],
-                                                                   'to': [metrics['to'] if 'to' in metrics else None],
-                                                                   })])
-    except Exception as e:
-        print(f"Exception: {type(e).__name__} - {e}")
 
 
-def hybrid_metrics_watcher(metric_name: str, value: Union[int, float]) -> None:
+def hybrid_metrics_watcher(metric_name: str, metric_value: Union[int, float]) -> None:
+    global hybrid_switch_log
+    mode = None
     if metric_name.startswith("modin.hybrid.auto"):
-        tokens = metric_name.split(".")
-        from_engine = None
-        to_engine = None
-        metric = None
-        group = None
-        submetric = None
-        if len(tokens) >= 9:
-            from_engine = tokens[4]
-            to_engine = tokens[6]
-            metric = tokens[7]
-        if len(tokens) == 9:
-            group = tokens[8]
-        if len(tokens) == 10:
-            submetric = tokens[8]
-            group = tokens[9]
-        add_to_hybrid_switch_log({'mode': 'single', 
-                                  'from': from_engine, 
-                                  'to': to_engine, 
-                                  'metric': metric, 
-                                  'submetric': submetric, 
-                                  'group': group, 
-                                  'value': value})
-    if metric_name.startswith("modin.hybrid.cast"):
-        tokens = metric_name.split(".")
-        to_engine = None
-        metric = None
-        submetric = None
-        group = None
-        if len(tokens) == 7 and tokens[3] == 'to' and tokens[5] == 'cost':
-            to_engine = tokens[4]
-            group = tokens[6]
-            metric = 'cost'
-        if len(tokens) == 6 and tokens[3] == 'decision':
-            submetric = tokens[4]
-            group = tokens[5]
-            metric = 'decision'
-        add_to_hybrid_switch_log({'mode': 'merge', 
-                                  'to': to_engine, 
-                                  'metric': metric,
-                                  'submetric': submetric,
-                                  'group': group, 
-                                  'value': value})
-        
+        mode = "auto"
+    elif metric_name.startswith("modin.hybrid.merge"):
+        mode = "merge"
+    else:
+        return
+    tokens = metric_name.split(".")[3:]
+    entry = {'mode': mode}
+    while len(tokens) >= 2:
+        key = tokens.pop(0)
+        if key == 'api':
+            value = tokens.pop(0) + '.' + tokens.pop(0)
+        else:
+            value = tokens.pop(0)
+        entry[key] = value
 
-
-
+    if len(tokens) == 1:
+        key = tokens.pop(0)
+        entry[key] = metric_value
+    
+    source = get_user_source_location(entry['group'])
+    entry['source'] = source['source']
+    new_row = native_pd.DataFrame(entry, index=[0])
+    hybrid_switch_log = native_pd.concat([hybrid_switch_log, new_row])
+ 
 def connect_modin_telemetry() -> None:
     MetricsMode.enable()
     add_metric_handler(snowpark_pandas_api_watcher)

src/snowflake/snowpark/modin/plugin/extensions/pd_extensions.py

@@ -39,10 +39,16 @@ def _snowpark_pandas_obj_check(obj: Union[DataFrame, Series]):
 
 @register_pd_accessor("explain")
 def explain(last=20) -> native_pd.DataFrame:
-    stats = get_hybrid_switch_log().tail(last)
-    stats = stats.drop_duplicates().fillna(value=' ')
-    stats2 = stats.reset_index()
-    return stats2.set_index([ 'source', 'mode', 'from', 'to', 'metric']).drop(columns=['index', 'group'])
+    stats = get_hybrid_switch_log()
+    stats = stats.reset_index(drop=True).sort_index().reset_index()
+    stats['decision'] = stats.groupby(['group']).bfill().ffill()['decision']
+    stats['api'] = stats.groupby(['group']).bfill().ffill()['api']
+    stats = stats.groupby('group', sort=False).apply(include_groups=False,func=lambda x: x.melt(ignore_index=False, id_vars=['source', 'api', 'decision', 'candidate', 'index', 'mode'], var_name='metric', value_vars=['stay_cost', 'move_to_cost', 'other_execute_cost', 'cost', 'delta', 'rows', 'cols'])).dropna()
+    stats = stats.set_index(['source', 'decision', 'api', 'index'])
+    stats = stats.sort_index(level='index')
+    stats['value'] = stats['value'].astype(int)
+    stats.reset_index().drop(columns='index').set_index(['source', 'decision', 'api'])
+    return stats.tail(last)
 
 @register_pd_accessor("read_snowflake")
 def read_snowflake(