refine

msr.py

@@ -30,6 +30,10 @@ LEADERBOARD_TIME_FRAME_DAYS = 180  # Time frame for leaderboard
 GHARCHIVE_DATA_DIR = "../gharchive/data"  # Local GHArchive data directory
 DUCKDB_CACHE_FILE = "../gharchive/gharchive_cache.duckdb"  # Persistent DuckDB database for caching
 
+# DuckDB performance configuration
+DUCKDB_THREADS = 8  # Number of threads for parallel processing
+DUCKDB_MEMORY_LIMIT = "64GB"  # Memory limit to prevent OOM crashes
+
 # Download configuration
 DOWNLOAD_WORKERS = 4  # Number of parallel download threads
 DOWNLOAD_RETRY_DELAY = 2  # Initial retry delay in seconds
@@ -332,11 +336,13 @@ def get_duckdb_connection():
     # Use persistent database for caching results
     conn = duckdb.connect(DUCKDB_CACHE_FILE)
 
-    # Optimize for 96-core CPU parallelization with 754GB RAM
-    conn.execute("SET threads TO 8;")  # Use all available cores
+    # Optimize for parallel processing with memory limits
+    conn.execute(f"SET threads TO {DUCKDB_THREADS};")  # Configure parallel threads
     conn.execute("SET preserve_insertion_order = false;")  # Better parallelization
     conn.execute("SET enable_object_cache = true;")  # Cache objects for reuse
     conn.execute("SET temp_directory = '/tmp/duckdb_temp';")  # Use fast temp storage if needed
+    conn.execute(f"SET memory_limit = '{DUCKDB_MEMORY_LIMIT}';")  # Limit memory to prevent OOM crashes
+    conn.execute(f"SET max_memory = '{DUCKDB_MEMORY_LIMIT}';")  # Hard memory cap
 
     return conn
 
@@ -428,54 +434,51 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
     identifier_list = ', '.join([f"'{id}'" for id in identifiers])
 
     # Build comprehensive query with CTEs using parameterized file lists (JSON.gz format)
+    # Optimized: Single file scan + ROW_NUMBER() deduplication (no DISTINCT)
     query = f"""
-    WITH review_events AS (
-      -- Get all review events for ALL agents
+    WITH all_review_events AS (
+      -- Single file scan for all three event types (optimization: 3x I/O reduction)
       SELECT
-        TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR) as url,
-        COALESCE(
-          TRY_CAST(json_extract_string(payload, '$.review.submitted_at') AS VARCHAR),
-          TRY_CAST(created_at AS VARCHAR)
-        ) as reviewed_at,
+        TRY_CAST(type AS VARCHAR) as event_type,
         TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
         TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
-        TRY_CAST(json_extract_string(payload, '$.pull_request.number') AS INTEGER) as pr_number
+        payload,
+        created_at
       FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
       WHERE
-        TRY_CAST(type AS VARCHAR) = 'PullRequestReviewEvent'
+        TRY_CAST(type AS VARCHAR) IN ('PullRequestReviewEvent', 'IssueCommentEvent', 'PullRequestReviewCommentEvent')
         AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
-        AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL
-
-      UNION ALL
-
-      -- Get PR comments (IssueCommentEvent on PRs)
-      SELECT
-        TRY_CAST(json_extract_string(payload, '$.issue.html_url') AS VARCHAR) as url,
-        TRY_CAST(created_at AS VARCHAR) as reviewed_at,
-        TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
-        TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
-        TRY_CAST(json_extract_string(payload, '$.issue.number') AS INTEGER) as pr_number
-      FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
-      WHERE
-        TRY_CAST(type AS VARCHAR) = 'IssueCommentEvent'
-        AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
-        AND json_extract_string(payload, '$.issue.pull_request.url') IS NOT NULL
-        AND json_extract_string(payload, '$.issue.html_url') IS NOT NULL
-
-      UNION ALL
+    ),
 
-      -- Get review comments (PullRequestReviewCommentEvent)
+    review_events AS (
+      -- Process events conditionally based on type
       SELECT
-        TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR) as url,
-        TRY_CAST(created_at AS VARCHAR) as reviewed_at,
-        TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
-        TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
-        TRY_CAST(json_extract_string(payload, '$.pull_request.number') AS INTEGER) as pr_number
-      FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
+        CASE
+          WHEN event_type = 'IssueCommentEvent'
+            THEN TRY_CAST(json_extract_string(payload, '$.issue.html_url') AS VARCHAR)
+          ELSE TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR)
+        END as url,
+        CASE
+          WHEN event_type = 'PullRequestReviewEvent'
+            THEN COALESCE(
+              TRY_CAST(json_extract_string(payload, '$.review.submitted_at') AS VARCHAR),
+              TRY_CAST(created_at AS VARCHAR)
+            )
+          ELSE TRY_CAST(created_at AS VARCHAR)
+        END as reviewed_at,
+        reviewer,
+        repo_name,
+        CASE
+          WHEN event_type = 'IssueCommentEvent'
+            THEN TRY_CAST(json_extract_string(payload, '$.issue.number') AS INTEGER)
+          ELSE TRY_CAST(json_extract_string(payload, '$.pull_request.number') AS INTEGER)
+        END as pr_number
+      FROM all_review_events
       WHERE
-        TRY_CAST(type AS VARCHAR) = 'PullRequestReviewCommentEvent'
-        AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
-        AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL
+        -- Validate required fields per event type
+        (event_type = 'PullRequestReviewEvent' AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL)
+        OR (event_type = 'IssueCommentEvent' AND json_extract_string(payload, '$.issue.pull_request.url') IS NOT NULL AND json_extract_string(payload, '$.issue.html_url') IS NOT NULL)
+        OR (event_type = 'PullRequestReviewCommentEvent' AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL)
     ),
 
     pr_status AS (
@@ -495,18 +498,34 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
         AND json_extract_string(payload, '$.pull_request.html_url') IN (
           SELECT DISTINCT url FROM review_events
         )
+    ),
+
+    deduplicated_reviews AS (
+      -- Efficient deduplication using ROW_NUMBER() instead of DISTINCT (optimization: prevents massive hash table)
+      SELECT
+        re.reviewer,
+        re.url,
+        re.reviewed_at,
+        ps.merged_at,
+        ps.closed_at,
+        ROW_NUMBER() OVER (
+          PARTITION BY re.reviewer, re.url, re.reviewed_at
+          ORDER BY re.reviewed_at
+        ) as row_num
+      FROM review_events re
+      LEFT JOIN (SELECT * FROM pr_status WHERE rn = 1) ps ON re.url = ps.url
     )
 
-    -- Join review events with PR status
-    SELECT DISTINCT
-      re.reviewer,
-      re.url,
-      re.reviewed_at,
-      ps.merged_at,
-      ps.closed_at
-    FROM review_events re
-    LEFT JOIN (SELECT * FROM pr_status WHERE rn = 1) ps ON re.url = ps.url
-    ORDER BY re.reviewer, re.reviewed_at DESC
+    -- Return deduplicated results (row_num = 1 ensures uniqueness without DISTINCT)
+    SELECT
+      reviewer,
+      url,
+      reviewed_at,
+      merged_at,
+      closed_at
+    FROM deduplicated_reviews
+    WHERE row_num = 1
+    ORDER BY reviewer, reviewed_at DESC
     """
 
     try:
@@ -546,8 +565,10 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
                     [r[4] for r in results]
                 ])
 
-        # Group results by agent
+        # Group results by agent with verification
         metadata_by_agent = defaultdict(list)
+        unique_reviews = set()
+        duplicate_count = 0
 
         for row in results:
             reviewer = row[0]
@@ -556,6 +577,12 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
             merged_at = normalize_date_format(row[3]) if row[3] else None
             closed_at = normalize_date_format(row[4]) if row[4] else None
 
+            # Track unique review combinations for verification
+            review_key = (reviewer, url, reviewed_at)
+            if review_key in unique_reviews:
+                duplicate_count += 1
+            unique_reviews.add(review_key)
+
             metadata_by_agent[reviewer].append({
                 'url': url,
                 'reviewed_at': reviewed_at,
@@ -563,6 +590,14 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
                 'closed_at': closed_at,
             })
 
+        # Verification: Ensure we have unique reviews (no duplicates from query)
+        total_reviews = len(results)
+        if duplicate_count > 0:
+            print(f"   Warning: Found {duplicate_count} duplicate review entries in query results!")
+            print(f"   Total: {total_reviews}, Unique: {len(unique_reviews)}")
+        else:
+            print(f"   Verification passed: {len(unique_reviews)} unique reviews retrieved (no duplicates)")
+
         # Convert defaultdict to regular dict
         return dict(metadata_by_agent)