@Override
  public Set<QueryId> getInsertQueries(
      final SourceName sourceName,
      final BiPredicate<SourceName, PersistentQueryMetadata> filterQueries) {
    return insertQueries.getOrDefault(sourceName, Collections.emptySet()).stream()
        .map(persistentQueries::get)
        .filter(query -> filterQueries.test(sourceName, query))
        .map(QueryMetadata::getQueryId)
        .collect(Collectors.toSet());
  }
@Test
  public void shouldGetQueriesInsertingIntoOrReadingFromSource() {
    givenCreate(registry, "q1", "source", Optional.of("sink1"), CREATE_AS);
    givenCreate(registry, "q2", "source", Optional.of("sink1"), INSERT);
    givenCreate(registry, "q3", "source", Optional.of("sink1"), INSERT);
    givenCreate(registry, "q4", "sink1", Optional.of("sink2"), INSERT);

    // When:
    final Set<QueryId> queries = registry.getInsertQueries(SourceName.of("sink1"), (n, q) -> true);

    // Then:
    assertThat(queries, contains(new QueryId("q2"), new QueryId("q3"), new QueryId("q4")));
  }