<commit_before>package com.socrata.datacoordinator.secondary.sql

import java.sql.Connection

import com.rojoma.simplearm.util.using
import com.socrata.datacoordinator.secondary.CollocationManifest

abstract class SqlCollocationManifest(conn: Connection) extends CollocationManifest {
  override def collocatedDatasets(datasets: Set[String]) = {
    using(conn.prepareStatement(
      """WITH RECURSIVE related_data_sets(dataset_internal_name_left, dataset_internal_name_right, system_id, path, cycle) AS (
        |    SELECT dataset_internal_name_left, dataset_internal_name_right, system_id, ARRAY[system_id], false
        |      FROM collocation_manifest WHERE dataset_internal_name_left = ANY(?) OR dataset_internal_name_right = ANY(?)
        |    UNION ALL
        |    SELECT src.dataset_internal_name_left, src.dataset_internal_name_right, src.system_id, path || src.system_id, src.system_id = ANY(path)
        |      FROM collocation_manifest src, related_data_sets rn
        |      WHERE (
        |          src.dataset_internal_name_left=rn.dataset_internal_name_right OR
        |          src.dataset_internal_name_right=rn.dataset_internal_name_left OR
        |          src.dataset_internal_name_left=rn.dataset_internal_name_left OR
        |          src.dataset_internal_name_right=rn.dataset_internal_name_right
        |        )
        |        AND NOT cycle
        |      )
        |SELECT dataset_internal_name_left FROM related_data_sets
        |UNION
        |SELECT dataset_internal_name_right FROM related_data_sets""".stripMargin))
    { stmt =>
      val datasetsArray =  conn.createArrayOf("varchar", datasets.toArray)
      stmt.setArray(1, datasetsArray)
      stmt.setArray(2, datasetsArray)
      using(stmt.executeQuery()) { rs =>
        val result = Set.newBuilder[String]
        while (rs.next()) {
          result += rs.getString("dataset_internal_name_left")
        }
        result.result()
      }
    }
  }

  override def dropCollocations(dataset: String): Unit = {
    using(conn.prepareStatement(
      """DELETE FROM collocation_manifest
        | WHERE dataset_internal_name_left = ?
        |    OR dataset_internal_name_right = ?
      """.stripMargin)) { stmt =><commit_msg>EN-24804: Fix collocation manifest recursive query (#244)<commit_after>package com.socrata.datacoordinator.secondary.sql

import java.sql.Connection

import com.rojoma.simplearm.util.using
import com.socrata.datacoordinator.secondary.CollocationManifest

import scala.collection.mutable

abstract class SqlCollocationManifest(conn: Connection) extends CollocationManifest {
  override def collocatedDatasets(datasets: Set[String]): Set[String] = {
    def getNeighbors(dataset: String): Set[String] = {
      using(conn.prepareStatement(
        """SELECT dataset_internal_name_left FROM collocation_manifest WHERE dataset_internal_name_right = ?
          |UNION
          |SELECT dataset_internal_name_right FROM collocation_manifest WHERE dataset_internal_name_left = ?""".stripMargin))
      { stmt =>
        stmt.setString(1, dataset)
        stmt.setString(2, dataset)
        using(stmt.executeQuery()) { rs =>
          val result = Set.newBuilder[String]
          while (rs.next()) {
            result += rs.getString("dataset_internal_name_left")
          }
          result.result()
        }
      }
    }

    val toVisit = mutable.Queue.empty[String]
    val visited = mutable.Set.empty[String]

    datasets.foreach(toVisit.enqueue(_))

    while (toVisit.nonEmpty) {
      val current = toVisit.dequeue()
      visited.add(current)

      getNeighbors(current).foreach { neighbor =>
        if (!visited.contains(neighbor) && !toVisit.contains(neighbor)) toVisit.enqueue(neighbor)
      }
    }

    visited.toSet
  }

  override def dropCollocations(dataset: String): Unit = {
    using(conn.prepareStatement(
      """DELETE FROM collocation_manifest
        | WHERE dataset_internal_name_left = ?
        |    OR dataset_internal_name_right = ?
      """.stripMargin)) { stmt =>
{'fraction_non_alphanumeric': 0.09187716691431402, 'fraction_numerical': 0.0029717682020802376, 'mean_word_length': 3.207291666666667, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5049185', 'n_tokens_mistral': 1193, 'n_tokens_neox': 1168, 'n_words': 206}