Merge pull request #9 from YanivZalach/feat/reactor_backend

Feat/reactor backend

Author: YanivZalach

backend/iceberg_inventory_builder.py

@@ -1,11 +1,11 @@
+import threading
+import traceback
 from dotenv import load_dotenv
 from pathlib import Path
 from flask import (
     Flask,
     jsonify,
     request,
-    Response,
-    redirect,
     send_from_directory,
 )
 from pyspark.errors import AnalysisException
@@ -19,6 +19,9 @@
 load_dotenv()
 app = Flask(__name__, static_url_path="/static")
 
+_in_flight_lock = threading.Lock()
+_in_flight: dict = {}
+
 
 @app.route("/", defaults={"path": ""})
 @app.route("/<path:path>")
@@ -36,18 +39,45 @@ def serve(path):
 def graph_data():
     table_name = request.form.get("table_name")
     date_value = request.form.get("date")
+    key = (table_name, date_value)
+
+    with _in_flight_lock:
+        if key in _in_flight:
+            state = _in_flight[key]
+            is_leader = False
+        else:
+            state = {"event": threading.Event(), "result": None, "error": None}
+            _in_flight[key] = state
+            is_leader = True
+
+    if not is_leader:
+        logger.info(f"Duplicate request for {key}, waiting for in-flight result")
+        state["event"].wait()
+        if state["error"]:
+            return jsonify({"error": state["error"]}), 400
+        return jsonify(state["result"])
 
     try:
         verify_iceberg_table(table_name)
         table_data = IcebergInventoryBuilder(table_name, date_value).collect()
-        data = normalize_graph_data(table_data)
-
-        return jsonify(data)
+        state["result"] = normalize_graph_data(table_data)
+        return jsonify(state["result"])
 
     except AnalysisException as e:
-        logger.error(f"Spark Error: {e}")
+        logger.error(f"Spark Error: {e}\n{traceback.format_exc()}")
+        state["error"] = str(e)
         return jsonify({"error": str(e)}), 400
 
+    except Exception as e:
+        logger.error(f"Unexpected error: {e}\n{traceback.format_exc()}")
+        state["error"] = str(e)
+        return jsonify({"error": str(e)}), 500
+
+    finally:
+        state["event"].set()
+        with _in_flight_lock:
+            _in_flight.pop(key, None)
+
 
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=APPLICATION_PORT)

backend/main.py

@@ -1,5 +1,6 @@
 from typing import List
 import json
+from pyspark.sql import functions as F
 import os
 from contextlib import suppress
 from datetime import datetime
@@ -126,6 +127,35 @@ def format_node_info(file_info: Dict[str, Any]) -> str:
     return formatted_info
 
 
+def get_metadata_row_slim_df_from_path(metadata_path: str):
+    spark = SparkSession.builder.getOrCreate()
+    df = spark.read.option("multiLine", True).json(metadata_path)
+    existing = set(df.columns)
+
+    scalar_cols = [
+        "current-schema-id",
+        "current-snapshot-id",
+        "default-sort-order-id",
+        "default-spec-id",
+        "last-column-id",
+        "last-partition-id",
+        "last-sequence-number",
+        "last-updated-ms",
+        "location",
+        "table-uuid",
+    ]
+    json_cols = ["properties", "refs"]
+
+    return df.select(
+        *[F.col(column) for column in scalar_cols if column in existing],
+        *[
+            F.to_json(F.col(column)).alias(column)
+            for column in json_cols
+            if column in existing
+        ],
+    )
+
+
 def get_json_metadata_from_path(metadata_path: str) -> Dict[str, Any]:
     spark = SparkSession.builder.getOrCreate()
 
@@ -142,7 +172,7 @@ def get_json_metadata_from_path(metadata_path: str) -> Dict[str, Any]:
     return row.asDict(recursive=True)
 
 
-def _update_col_metric(source_list, metric_name, column_metrics):
+def update_col_metric(source_list, metric_name, column_metrics):
     for row in source_list:
         col_id = row.key
         if col_id not in column_metrics:

backend/utils.py

@@ -9,8 +9,7 @@
 # ─────────────────────────────────────────────
 
 spark.sql("DROP TABLE IF EXISTS default.events")
-spark.sql(
-    """
+spark.sql("""
     CREATE TABLE default.events (
         event_id    INT,
         event_name  STRING,
@@ -24,8 +23,7 @@
         'write.update.mode' = 'merge-on-read',
         'write.merge.mode' = 'merge-on-read'
     )
-    """
-)
+    """)
 print("✅ Created table `default.events`, partitioned by hour(event_ts)\n")
 
 # ─────────────────────────────────────────────
@@ -175,8 +173,7 @@
 # ─────────────────────────────────────────────
 
 spark.sql("DROP TABLE IF EXISTS default.logging")
-spark.sql(
-    """
+spark.sql("""
     CREATE TABLE default.logging (
         event_id    INT,
         event_name  STRING,
@@ -190,8 +187,7 @@
         'write.update.mode' = 'merge-on-read',
         'write.merge.mode' = 'merge-on-read'
     )
-    """
-)
+    """)
 print("✅ Created table `default.logging`, partitioned by hour(event_ts)\n")
 
 # ─────────────────────────────────────────────

docker_demo/create_mock_tables.py

@@ -323,6 +323,12 @@ export default function FileTreePage() {
       .sort((a, b) => a.name.localeCompare(b.name))
   }, [metadata])
 
+  useEffect(() => {
+    if (selectedBranch !== null) return
+    const mainBranch = branches.find(b => b.name === 'main')
+    if (mainBranch) setSelectedBranch('main')
+  }, [branches])
+
   const displayedSnapshots = useMemo(() => {
     if (!selectedBranch) return snapshots
     const branch = branches.find(b => b.name === selectedBranch)

frontend/src/mocks/handlers.js

@@ -68,6 +68,7 @@ export default function GraphPage() {
     liveEdges.update(liveEdges.get().map(e => ({ ...e, hidden: false })))
     setStickyNode(null)
     setIsFullView(true)
+    history.replaceState({ graphSelection: null }, '')
     requestAnimationFrame(() => { network.redraw(); network.fit() })
   }, [])
 
@@ -232,16 +233,31 @@ export default function GraphPage() {
                 🔒 Locked View
               </span>
             )}
-            {sticky.rows.map((r, i) => (
-              <div key={i}>
-                <span className="block font-bold text-slate-500 text-[0.65rem] uppercase tracking-wider mb-1">
-                  {r.label}
-                </span>
-                <span className="block font-mono bg-[#0d1117] text-slate-200 px-3 py-2 rounded-lg text-xs whitespace-pre overflow-x-auto break-normal">
-                  {r.value}
-                </span>
-              </div>
-            ))}
+            {sticky.rows.map((r, i) => {
+              let displayValue = r.value
+              const tryParseJson = (str) => {
+                try { return JSON.parse(str) } catch { return undefined }
+              }
+              const asPythonToJson = (str) => str
+                .replace(/'/g, '"')
+                .replace(/\bTrue\b/g, 'true')
+                .replace(/\bFalse\b/g, 'false')
+                .replace(/\bNone\b/g, 'null')
+              const parsed = tryParseJson(r.value) ?? tryParseJson(asPythonToJson(r.value))
+              if (parsed !== undefined && typeof parsed === 'object' && parsed !== null) {
+                displayValue = JSON.stringify(parsed, null, 2)
+              }
+              return (
+                <div key={i}>
+                  <span className="block font-bold text-slate-500 text-[0.65rem] uppercase tracking-wider mb-1">
+                    {r.label}
+                  </span>
+                  <span className="block font-mono bg-[#0d1117] text-slate-200 px-3 py-2 rounded-lg text-xs whitespace-pre overflow-x-auto break-normal">
+                    {displayValue}
+                  </span>
+                </div>
+              )
+            })}
           </div>
         </div>
       )}

frontend/src/pages/FileTreePage.jsx

@@ -78,15 +78,20 @@ export default function TableLayout() {
     }
 
     const body = new URLSearchParams({ table_name: tableName, date })
+    const controller = new AbortController()
 
     fetch('/api/v1/graph-data', {
       method: 'POST',
       headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
       body: body.toString(),
+      signal: controller.signal,
     })
       .then(async (res) => {
         const text = await res.text()
         const data = JSONbig({ storeAsString: true }).parse(text)
+
+        console.log(data)
+
         if (!res.ok) throw new Error(data.error || 'Request failed')
         sessionStorage.setItem(cacheKey, text)
         return data
@@ -96,9 +101,12 @@ export default function TableLayout() {
         setLoading(false)
       })
       .catch((err) => {
+        if (err.name === 'AbortError') return
         setError(err.message || 'An unexpected error occurred.')
         setLoading(false)
       })
+
+    return () => controller.abort()
   }, [tableName, date])
 
   if (loading) {