Deduplicate Hub models and datasets before import

database_refresh/refresh_database.py

@@ -66,16 +66,36 @@ def create_views(
     connection.execute(
         f"""
         CREATE VIEW source_models AS
-        SELECT *
-        FROM read_parquet('{sql_path(models_path)}')
+        SELECT * EXCLUDE (_dedupe_rank)
+        FROM (
+            SELECT
+                *,
+                row_number() OVER (
+                    PARTITION BY id
+                    ORDER BY lastModified DESC NULLS LAST, _id DESC NULLS LAST
+                ) AS _dedupe_rank
+            FROM read_parquet('{sql_path(models_path)}')
+            WHERE id IS NOT NULL AND trim(id) <> ''
+        )
+        WHERE _dedupe_rank = 1
         {model_limit}
         """
     )
     connection.execute(
         f"""
         CREATE VIEW source_datasets AS
-        SELECT *
-        FROM read_parquet('{sql_path(datasets_path)}')
+        SELECT * EXCLUDE (_dedupe_rank)
+        FROM (
+            SELECT
+                *,
+                row_number() OVER (
+                    PARTITION BY id
+                    ORDER BY lastModified DESC NULLS LAST, _id DESC NULLS LAST
+                ) AS _dedupe_rank
+            FROM read_parquet('{sql_path(datasets_path)}')
+            WHERE id IS NOT NULL AND trim(id) <> ''
+        )
+        WHERE _dedupe_rank = 1
         {dataset_limit}
         """
     )
@@ -335,10 +355,14 @@ def write_metadata(
     return path
 
 
-def parquet_count(path: Path) -> int:
+def parquet_unique_id_count(path: Path) -> int:
     connection = duckdb.connect()
     count = connection.execute(
-        f"SELECT count(*) FROM read_parquet('{sql_path(path)}')"
+        f"""
+        SELECT count(DISTINCT id)
+        FROM read_parquet('{sql_path(path)}')
+        WHERE id IS NOT NULL AND trim(id) <> ''
+        """
     ).fetchone()[0]
     connection.close()
     return int(count)
@@ -428,8 +452,8 @@ def main() -> int:
         return 0
 
     dump_path = build_dump(files, args.work_dir, args.neo4j_admin)
-    model_count = args.max_models or parquet_count(models_path)
-    dataset_count = args.max_datasets or parquet_count(datasets_path)
+    model_count = args.max_models or parquet_unique_id_count(models_path)
+    dataset_count = args.max_datasets or parquet_unique_id_count(datasets_path)
     metadata_path = write_metadata(
         args.work_dir,
         source_revision,