fix query to get most recent actor catalog fetch event for some sources

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java

@@ -1355,12 +1355,13 @@ public Map<UUID, ActorCatalogFetchEvent> getMostRecentActorCatalogFetchEventForS
     // noinspection SqlResolve
     return database.query(ctx -> ctx.fetch(
         """
-        select actor_catalog_id, actor_id, created_at from
-          (select actor_catalog_id, actor_id, created_at, rank() over (partition by actor_id order by created_at desc) as creation_order_rank
+        select distinct actor_catalog_id, actor_id, created_at from
+          (select actor_catalog_id, actor_id, created_at, row_number() over (partition by actor_id order by created_at desc) as creation_order_row_number
           from public.actor_catalog_fetch_event
+          where actor_id in ({0})
           ) table_with_rank
-        where creation_order_rank = 1;
-        """))
+        where creation_order_row_number = 1;
+        """, DSL.list(sourceIds.stream().map(DSL::value).collect(Collectors.toList()))))
         .stream().map(DbConverter::buildActorCatalogFetchEvent)
         .collect(Collectors.toMap(ActorCatalogFetchEvent::getActorId, record -> record));
   }

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConfigRepositoryE2EReadWriteTest.java

@@ -598,6 +598,39 @@ void testGetMostRecentActorCatalogFetchEventForSources() throws SQLException, IO
     assertEquals(MockData.ACTOR_CATALOG_ID_3, result.get(MockData.SOURCE_ID_2).getActorCatalogId());
   }
 
+  @Test
+  void testGetMostRecentActorCatalogFetchEventWithDuplicates() throws SQLException, IOException {
+    // Tests that we can handle two fetch events in the db with the same actor id, actor catalog id, and
+    // timestamp e.g., from duplicate discoveries.
+    for (final ActorCatalog actorCatalog : MockData.actorCatalogs()) {
+      writeActorCatalog(database, Collections.singletonList(actorCatalog));
+    }
+
+    database.transaction(ctx -> {
+      // Insert the fetch events twice.
+      MockData.actorCatalogFetchEventsForAggregationTest().forEach(actorCatalogFetchEvent -> {
+        insertCatalogFetchEvent(
+            ctx,
+            actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorId(),
+            actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorCatalogId(),
+            actorCatalogFetchEvent.getCreatedAt());
+        insertCatalogFetchEvent(
+            ctx,
+            actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorId(),
+            actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorCatalogId(),
+            actorCatalogFetchEvent.getCreatedAt());
+      });
+      return null;
+    });
+
+    final Map<UUID, ActorCatalogFetchEvent> result =
+        configRepository.getMostRecentActorCatalogFetchEventForSources(List.of(MockData.SOURCE_ID_1,
+            MockData.SOURCE_ID_2));
+
+    assertEquals(MockData.ACTOR_CATALOG_ID_1, result.get(MockData.SOURCE_ID_1).getActorCatalogId());
+    assertEquals(MockData.ACTOR_CATALOG_ID_3, result.get(MockData.SOURCE_ID_2).getActorCatalogId());
+  }
+
   @Test
   void testGetActorDefinitionsInUseToProtocolVersion() throws IOException {
     final Set<UUID> actorDefinitionIds = new HashSet<>();

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java

@@ -691,10 +691,17 @@ public static List<ActorCatalogFetchEventWithCreationDate> actorCatalogFetchEven
         .withActorId(SOURCE_ID_2)
         .withConfigHash("1394")
         .withConnectorVersion("1.2.0");
+    final ActorCatalogFetchEvent actorCatalogFetchEvent4 = new ActorCatalogFetchEvent()
+        .withId(ACTOR_CATALOG_FETCH_EVENT_ID_3)
+        .withActorCatalogId(ACTOR_CATALOG_ID_3)
+        .withActorId(SOURCE_ID_3)
+        .withConfigHash("1394")
+        .withConnectorVersion("1.2.0");
     return Arrays.asList(
         new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent1, now),
         new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent2, yesterday),
-        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent3, now));
+        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent3, now),
+        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent4, now));
   }
 
   public static List<WorkspaceServiceAccount> workspaceServiceAccounts() {